提交工作

如要將工作提交至現有的 Managed Service for Apache Spark 叢集，您可以發出 jobs.submit API HTTP 或程式輔助要求、在本機終端機視窗或 Cloud Shell 使用 Google Cloud CLI gcloud 指令列工具，或是透過本機瀏覽器開啟的 Google Cloud 控制台操作。您也可以使用 SSH 連結叢集中的主要執行個體，然後直接從執行個體執行工作，無需使用 Managed Service for Apache Spark。

工作驅動程式和記錄輸出：如要瞭解如何查看工作驅動程式輸出和記錄，請參閱「Managed Service for Apache Spark 工作輸出內容和記錄」。

工作並行： 建立叢集時，您可以使用 dataproc:dataproc.scheduler.max-concurrent-jobs 叢集屬性，設定並行 Managed Service for Apache Spark 工作數量上限。如果未設定這個屬性值，並行工作的上限會計算為 max((masterMemoryMb - 3584) / masterMemoryMbPerJob, 5)。masterMemoryMb 取決於主要 VM 的機型。masterMemoryMbPerJob 預設為 1024，但可在建立叢集時，使用 dataproc:dataproc.scheduler.driver-size-mb 叢集屬性設定。

注意： 如果同時提交多項工作，可能會發生競爭狀況，導致工作失敗。為避免工作失敗，請確保連續提交工作之間至少間隔一小段時間。

如何提交工作

您可以指定 file:/// 路徑，參照叢集主要節點中的本機檔案。請注意，file:/// 路徑通常只用於叢集上提供的 Managed Service for Apache Spark 預先安裝檔案。

類別路徑隔離和自訂 JAR：請勿將自訂「胖 JAR」或套裝組合 JAR (例如 Apache Iceberg 執行階段或 Google Cloud 套裝組合) 直接複製到叢集系統目錄 (例如 /usr/lib/spark/jars/)。將自訂 JAR 放在系統目錄中，會導致 Managed Service for Apache Spark 代理程式的類別路徑受到遞移依附元件 (例如 Guava 或 Hadoop 程式庫) 污染，進而與代理程式的內建程式庫發生衝突。執行作業管理作業 (例如取消作業) 時，這項衝突可能會導致類別解析錯誤 (例如 ClassNotFoundException 或 NoClassDefFoundError)，進而導致代理程式當機，並讓 YARN 應用程式成為孤立應用程式。

請改用下列其中一種支援的做法：

• 工作專屬依附元件：使用 Google Cloud CLI CLI 中的 --jars 旗標、 Google Cloud 控制台中的「Jar 檔案」欄位，或 API 中的 jarFileUris 欄位提交工作時，請指定 JAR 的 Cloud Storage 路徑。Spark 會將依附元件分配給工作驅動程式和執行器，不會汙染代理程式的類別路徑。
• 叢集範圍的依附元件：建立叢集時，請使用 spark:spark.jars 叢集屬性指定依附元件：

  --properties="spark:spark.jars=gs://YOUR_BUCKET/jar-1.jar,gs://YOUR_BUCKET/jar-2.jar"
  

  這會指示 Managed Service for Apache Spark 設定 /etc/spark/conf/spark-defaults.conf，自動將依附元件分配至所有工作的 Spark 驅動程式和執行器類別路徑，同時讓節點本機的 Managed Service for Apache Spark 代理程式保持乾淨隔離。

控制台

在瀏覽器中開啟 Google Cloud 控制台，然後前往 Managed Service for Apache Spark「Submit a job」(提交工作) 頁面。

Spark 工作範例

如要提交範例 Spark 工作，請依照下列說明填寫「Submit a job」(提交工作) 頁面中的欄位：

1. 從叢集清單選取您的「Cluster」(叢集) 名稱。
2. 將「Job type」(工作類型) 設為 Spark。
3. 將「Main class or jar」(主要類別或 jar) 設為 org.apache.spark.examples.SparkPi。
4. 將「Arguments」(引數) 設為單一引數 1000。
5. 在「Jar files」(Jar 檔案) (或 API jarFileUris 欄位) 中加入 file:///usr/lib/spark/examples/jars/spark-examples.jar：
   1. file:/// 表示 Hadoop LocalFileSystem 配置。Managed Service for Apache Spark 會在建立叢集時，將 /usr/lib/spark/examples/jars/spark-examples.jar 安裝在叢集的主要節點上。這個路徑僅用於 Managed Service for Apache Spark 提供的預先安裝範例。請勿將自訂 JAR 複製到系統目錄 (請參閱「如何提交工作」中的注意事項)。
   2. 或者，您也可以指定其中一個 JAR 的 Cloud Storage 路徑 (gs://your-bucket/your-jarfile.jar) 或 Hadoop 分散式檔案系統路徑 (hdfs://path-to-jar.jar)。（如果您使用 API 提交作業，請在 jarFileUris 欄位中指定此路徑。）

新增更多引數： 每個引數必須在不同的文字方塊中輸入。請按下 <Return> 鍵，為每一個額外引數開啟一個新的文字方塊。

按一下 [Submit] (提交) 以啟動工作。工作啟動後，即會加入「Jobs」[工作] 清單中。

按一下工作 ID 以開啟「Jobs」(工作) 頁面，您可以在此頁面查看工作的驅動程式輸出。因為這項工作產生的輸出行長度超過瀏覽器視窗的寬度，您可以勾選「Line wrapping」(換行) 方塊，在視圖中顯示所有輸出文字，就可以顯示計算後的 pi 結果。

您可以使用下方顯示的 gcloud dataproc jobs wait 指令，從指令列查看工作的驅動程式輸出 (詳情請參閱查看工作輸出內容 - GCLOUD 指令)。複製專案 ID 並貼上做為 --project 標記的值；複製工作 ID (顯示在工作清單上) 並貼上做為最後一個引數的值。

gcloud dataproc jobs wait job-id \
    --project=project-id \
    --region=region

以下是範例 SparkPi 作業的驅動程式輸出片段：

...
2015-06-25 23:27:23,810 INFO [dag-scheduler-event-loop]
scheduler.DAGScheduler (Logging.scala:logInfo(59)) - Stage 0 (reduce at
SparkPi.scala:35) finished in 21.169 s

2015-06-25 23:27:23,810 INFO [task-result-getter-3] cluster.YarnScheduler
(Logging.scala:logInfo(59)) - Removed TaskSet 0.0, whose tasks have all
completed, from pool

2015-06-25 23:27:23,819 INFO [main] scheduler.DAGScheduler
(Logging.scala:logInfo(59)) - Job 0 finished: reduce at SparkPi.scala:35,
took 21.674931 s

Pi is roughly 3.14189648
...
Job [c556b47a-4b46-4a94-9ba2-2dcee31167b2] finished successfully.

driverOutputUri:
gs://sample-staging-bucket/google-cloud-dataproc-metainfo/cfeaa033-749e-48b9-...
...

gcloud

如要將工作提交至 Managed Service for Apache Spark 叢集，請在本機的終端機視窗或 Cloud Shell 中執行 gcloud CLI gcloud dataproc jobs submit 指令。

gcloud dataproc jobs submit job-command \
    --cluster=cluster-name \
    --region=region \
    other dataproc-flags \
    -- job-args

您可以新增 --cluster-labels 旗標，指定一或多個叢集標籤。Managed Service for Apache Spark 會將工作提交至符合指定叢集標籤的叢集。

PySpark 工作提交範例

1. 列出位於 Cloud Storage 中可公開存取的 hello-world.py。

   gcloud storage cat gs://dataproc-examples/pyspark/hello-world/hello-world.py
   

   列出檔案：

   #!/usr/bin/python
   import pyspark
   sc = pyspark.SparkContext()
   rdd = sc.parallelize(['Hello,', 'world!'])
   words = sorted(rdd.collect())
   print(words)

2. 將 PySpark 工作提交至 Managed Service for Apache Spark。

   gcloud dataproc jobs submit pyspark \
       gs://dataproc-examples/pyspark/hello-world/hello-world.py \
       --cluster=cluster-name  \
       --region=region
   

   終端機輸出：

   Waiting for job output...
   …
   ['Hello,', 'world!']
   Job finished successfully.
   

Spark 工作提交範例

1. 執行預先安裝在 Managed Service for Apache Spark 叢集主要節點上的 SparkPi 範例。(file:///usr/lib/spark/examples/jars/spark-examples.jar 路徑僅適用於預先安裝的範例；如需自訂 JAR 依附元件，請參閱「如何提交工作」一文中的注意事項)。

   gcloud dataproc jobs submit spark \
       --cluster=cluster-name \
       --region=region \
       --class=org.apache.spark.examples.SparkPi \
       --jars=file:///usr/lib/spark/examples/jars/spark-examples.jar \
       -- 1000
   

   終端機輸出：

   Job [54825071-ae28-4c5b-85a5-58fae6a597d6] submitted.
   Waiting for job output…
   …
   Pi is roughly 3.14177148
   …
   Job finished successfully.
   …
   

   工作如何計算 Pi：Spark 工作會使用蒙特卡羅法估算 Pi 的值。這個方法會在座標面上產生 x,y 點，建立由一個單位正方形包圍圓形的模型。輸入引數 (1000) 決定要產生的 x,y 配對數，配對數越多，預估就越準確。過程中，系統會利用 Managed Service for Apache Spark worker 節點平行處理計算工作。詳情請參閱「Estimating Pi using the Monte Carlo Method」(使用蒙地卡羅方法估算圓周率)，以及 GitHub 上的 JavaSparkPi.java。

REST

本節說明如何使用 Managed Service for Apache Spark jobs.submit API，提交 Spark 工作來計算 pi 的近似值。

您可以在下方顯示的 API 要求中新增 clusterLabels 欄位，指定一或多個叢集標籤。Managed Service for Apache Spark 會將工作提交至符合指定叢集標籤的叢集 (詳情請參閱 jobs.submit API)。

使用任何要求資料之前，請先修改下列項目的值：

• project-id： Google Cloud 專案 ID
• region：叢集區域
• clusterName：叢集名稱

HTTP 方法和網址：

POST https://dataproc.googleapis.com/v1/projects/project-id/regions/region/jobs:submit

JSON 要求主體：

{
  "job": {
    "placement": {
      "clusterName": "cluster-name"
    },
    "sparkJob": {
      "args": [
        "1000"
      ],
      "mainClass": "org.apache.spark.examples.SparkPi",
      "jarFileUris": [
        "file:///usr/lib/spark/examples/jars/spark-examples.jar"
      ]
    }
  }
}

請展開以下其中一個選項，以傳送要求：

curl (Linux、macOS 或 Cloud Shell)

注意：下列指令假設您已執行 gcloud init 或 gcloud auth login，透過使用者帳戶登入 gcloud CLI，或已使用 Cloud Shell 自動登入 gcloud CLI。您可以執行 gcloud auth list，查看目前使用的帳戶。

將要求主體儲存在名為 request.json 的檔案中，然後執行下列指令：

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://dataproc.googleapis.com/v1/projects/project-id/regions/region/jobs:submit"

PowerShell (Windows)

注意：下列指令假設您已執行 gcloud init 或 gcloud auth login，透過使用者帳戶登入 gcloud CLI。您可以執行 gcloud auth list，查看目前使用的帳戶。

將要求主體儲存在名為 request.json 的檔案中，然後執行下列指令：

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method POST `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://dataproc.googleapis.com/v1/projects/project-id/regions/region/jobs:submit" | Select-Object -Expand Content

您應該會收到如下的 JSON 回覆：

{
  "reference": {
    "projectId": "project-id",
    "jobId": "job-id"
  },
  "placement": {
    "clusterName": "cluster-name",
    "clusterUuid": "cluster-Uuid"
  },
  "sparkJob": {
    "mainClass": "org.apache.spark.examples.SparkPi",
    "args": [
      "1000"
    ],
    "jarFileUris": [
      "file:///usr/lib/spark/examples/jars/spark-examples.jar"
    ]
  },
  "status": {
    "state": "PENDING",
    "stateStartTime": "2020-10-07T20:16:21.759Z"
  },
  "jobUuid": "job-Uuid"
}

注意：您可以在 Managed Service for Apache Spark Google Cloud 主控台的「Submit a job」(提交工作) 頁面底端按一下「Equivalent REST」(對等 REST) 連結，讓 Google Cloud 主控台建構一個對等的 API REST 要求，並在程式碼中使用此要求以提交工作至叢集。

Java

1. 安裝用戶端程式庫
2. 設定應用程式預設憑證
3. 執行程式碼
   請參閱「設定 Java 開發環境」。
   
   import com.google.api.gax.longrunning.OperationFuture;
   import com.google.cloud.dataproc.v1.Job;
   import com.google.cloud.dataproc.v1.JobControllerClient;
   import com.google.cloud.dataproc.v1.JobControllerSettings;
   import com.google.cloud.dataproc.v1.JobMetadata;
   import com.google.cloud.dataproc.v1.JobPlacement;
   import com.google.cloud.dataproc.v1.SparkJob;
   import com.google.cloud.storage.Blob;
   import com.google.cloud.storage.Storage;
   import com.google.cloud.storage.StorageOptions;
   import java.io.IOException;
   import java.util.concurrent.ExecutionException;
   import java.util.regex.Matcher;
   import java.util.regex.Pattern;
   
   public class SubmitJob {
   
     public static void submitJob() throws IOException, InterruptedException {
       // TODO(developer): Replace these variables before running the sample.
       String projectId = "your-project-id";
       String region = "your-project-region";
       String clusterName = "your-cluster-name";
       submitJob(projectId, region, clusterName);
     }
   
     public static void submitJob(String projectId, String region, String clusterName)
         throws IOException, InterruptedException {
       String myEndpoint = String.format("%s-dataproc.googleapis.com:443", region);
   
       // Configure the settings for the job controller client.
       JobControllerSettings jobControllerSettings =
           JobControllerSettings.newBuilder().setEndpoint(myEndpoint).build();
   
       // Create a job controller client with the configured settings. Using a try-with-resources
       // closes the client,
       // but this can also be done manually with the .close() method.
       try (JobControllerClient jobControllerClient =
           JobControllerClient.create(jobControllerSettings)) {
   
         // Configure cluster placement for the job.
         JobPlacement jobPlacement = JobPlacement.newBuilder().setClusterName(clusterName).build();
   
         // Configure Spark job settings.
         SparkJob sparkJob =
             SparkJob.newBuilder()
                 .setMainClass("org.apache.spark.examples.SparkPi")
                 .addJarFileUris("file:///usr/lib/spark/examples/jars/spark-examples.jar")
                 .addArgs("1000")
                 .build();
   
         Job job = Job.newBuilder().setPlacement(jobPlacement).setSparkJob(sparkJob).build();
   
         // Submit an asynchronous request to execute the job.
         OperationFuture<Job, JobMetadata> submitJobAsOperationAsyncRequest =
             jobControllerClient.submitJobAsOperationAsync(projectId, region, job);
   
         Job response = submitJobAsOperationAsyncRequest.get();
   
         // Print output from Google Cloud Storage.
         Matcher matches =
             Pattern.compile("gs://(.*?)/(.*)").matcher(response.getDriverOutputResourceUri());
         matches.matches();
   
         Storage storage = StorageOptions.getDefaultInstance().getService();
         Blob blob = storage.get(matches.group(1), String.format("%s.000000000", matches.group(2)));
   
         System.out.println(
             String.format("Job finished successfully: %s", new String(blob.getContent())));
   
       } catch (ExecutionException e) {
         // If the job does not complete successfully, print the error message.
         System.err.println(String.format("submitJob: %s ", e.getMessage()));
       }
     }
   }

Python

1. 安裝用戶端程式庫
2. 設定應用程式預設憑證
3. 執行程式碼
   請參閱「設定 Python 開發環境」。
   import re
   
   
   from google.cloud import dataproc_v1 as dataproc
   from google.cloud import storage
   
   
   def submit_job(project_id, region, cluster_name):
       # Create the job client.
       job_client = dataproc.JobControllerClient(
           client_options={"api_endpoint": f"{region}-dataproc.googleapis.com:443"}
       )
   
       # Create the job config. 'main_jar_file_uri' can also be a
       # Google Cloud Storage URL.
       job = {
           "placement": {"cluster_name": cluster_name},
           "spark_job": {
               "main_class": "org.apache.spark.examples.SparkPi",
               "jar_file_uris": ["file:///usr/lib/spark/examples/jars/spark-examples.jar"],
               "args": ["1000"],
           },
       }
   
       operation = job_client.submit_job_as_operation(
           request={"project_id": project_id, "region": region, "job": job}
       )
       response = operation.result()
   
       # Dataproc job output gets saved to the Google Cloud Storage bucket
       # allocated to the job. Use a regex to obtain the bucket and blob info.
       matches = re.match("gs://(.*?)/(.*)", response.driver_output_resource_uri)
   
       output = (
           storage.Client()
           .get_bucket(matches.group(1))
           .blob(f"{matches.group(2)}.000000000")
           .download_as_bytes()
           .decode("utf-8")
       )
   
       print(f"Job finished successfully: {output}")
   
   

Go

1. 安裝用戶端程式庫
2. 設定應用程式預設憑證
3. 執行程式碼
   請參閱「設定 Go 開發環境」。
   import (
   	"context"
   	"fmt"
   	"io"
   	"log"
   	"regexp"
   
   	dataproc "cloud.google.com/go/dataproc/apiv1"
   	"cloud.google.com/go/dataproc/apiv1/dataprocpb"
   	"cloud.google.com/go/storage"
   	"google.golang.org/api/option"
   )
   
   func submitJob(w io.Writer, projectID, region, clusterName string) error {
   	// projectID := "your-project-id"
   	// region := "us-central1"
   	// clusterName := "your-cluster"
   	ctx := context.Background()
   
   	// Create the job client.
   	endpoint := fmt.Sprintf("%s-dataproc.googleapis.com:443", region)
   	jobClient, err := dataproc.NewJobControllerClient(ctx, option.WithEndpoint(endpoint))
   	if err != nil {
   		log.Fatalf("error creating the job client: %s\n", err)
   	}
   
   	// Create the job config.
   	submitJobReq := &dataprocpb.SubmitJobRequest{
   		ProjectId: projectID,
   		Region:    region,
   		Job: &dataprocpb.Job{
   			Placement: &dataprocpb.JobPlacement{
   				ClusterName: clusterName,
   			},
   			TypeJob: &dataprocpb.Job_SparkJob{
   				SparkJob: &dataprocpb.SparkJob{
   					Driver: &dataprocpb.SparkJob_MainClass{
   						MainClass: "org.apache.spark.examples.SparkPi",
   					},
   					JarFileUris: []string{"file:///usr/lib/spark/examples/jars/spark-examples.jar"},
   					Args:        []string{"1000"},
   				},
   			},
   		},
   	}
   
   	submitJobOp, err := jobClient.SubmitJobAsOperation(ctx, submitJobReq)
   	if err != nil {
   		return fmt.Errorf("error with request to submitting job: %w", err)
   	}
   
   	submitJobResp, err := submitJobOp.Wait(ctx)
   	if err != nil {
   		return fmt.Errorf("error submitting job: %w", err)
   	}
   
   	re := regexp.MustCompile("gs://(.+?)/(.+)")
   	matches := re.FindStringSubmatch(submitJobResp.DriverOutputResourceUri)
   
   	if len(matches) < 3 {
   		return fmt.Errorf("regex error: %s", submitJobResp.DriverOutputResourceUri)
   	}
   
   	// Dataproc job output gets saved to a GCS bucket allocated to it.
   	storageClient, err := storage.NewClient(ctx)
   	if err != nil {
   		return fmt.Errorf("error creating storage client: %w", err)
   	}
   
   	obj := fmt.Sprintf("%s.000000000", matches[2])
   	reader, err := storageClient.Bucket(matches[1]).Object(obj).NewReader(ctx)
   	if err != nil {
   		return fmt.Errorf("error reading job output: %w", err)
   	}
   
   	defer reader.Close()
   
   	body, err := io.ReadAll(reader)
   	if err != nil {
   		return fmt.Errorf("could not read output from Dataproc Job: %w", err)
   	}
   
   	fmt.Fprintf(w, "Job finished successfully: %s", body)
   
   	return nil
   }
   

Node.js

1. 安裝用戶端程式庫
2. 設定應用程式預設憑證
3. 執行程式碼
   請參閱「設定 Node.js 開發環境」。
   const dataproc = require('@google-cloud/dataproc');
   const {Storage} = require('@google-cloud/storage');
   
   // TODO(developer): Uncomment and set the following variables
   // projectId = 'YOUR_PROJECT_ID'
   // region = 'YOUR_CLUSTER_REGION'
   // clusterName = 'YOUR_CLUSTER_NAME'
   
   // Create a client with the endpoint set to the desired cluster region
   const jobClient = new dataproc.v1.JobControllerClient({
     apiEndpoint: `${region}-dataproc.googleapis.com`,
     projectId: projectId,
   });
   
   async function submitJob() {
     const job = {
       projectId: projectId,
       region: region,
       job: {
         placement: {
           clusterName: clusterName,
         },
         sparkJob: {
           mainClass: 'org.apache.spark.examples.SparkPi',
           jarFileUris: [
             'file:///usr/lib/spark/examples/jars/spark-examples.jar',
           ],
           args: ['1000'],
         },
       },
     };
   
     const [jobOperation] = await jobClient.submitJobAsOperation(job);
     const [jobResponse] = await jobOperation.promise();
   
     const matches =
       jobResponse.driverOutputResourceUri.match('gs://(.*?)/(.*)');
   
     const storage = new Storage();
   
     const output = await storage
       .bucket(matches[1])
       .file(`${matches[2]}.000000000`)
       .download();
   
     // Output a success message.
     console.log(`Job finished successfully: ${output}`);

直接在叢集中提交工作

如果您想在叢集中直接執行工作而不使用 Managed Service for Apache Spark，請使用 SSH 連結叢集的主要節點，然後在主要節點上執行工作。

建立連結 VM 主要執行個體的 SSH 連線後，請在叢集主要節點的終端機視窗中執行指令以完成下列步驟：

1. 開啟 Spark 殼層。
2. 執行 Spark 作業，統計位於可公開存取的 Cloud Storage 檔案中的（七行）Python「hello-world」檔案的行數。

3. 結束殼層。

   user@cluster-name-m:~$ spark-shell
   ...
   scala> sc.textFile("gs://dataproc-examples"
   + "/pyspark/hello-world/hello-world.py").count
   ...
   res0: Long = 7
   scala> :quit
   

在 Managed Service for Apache Spark 上執行 Bash 工作

您可能想將 Bash 指令碼做為 Managed Service for Apache Spark 工作執行，原因可能是您使用的引擎不支援做為頂層 Managed Service for Apache Spark 工作類型，或是您需要先完成其他設定或計算引數，再從指令碼使用 hadoop 或 spark-submit 啟動工作。

Pig 範例

假設您已將 hello.sh bash 指令碼複製到 Cloud Storage：

gcloud storage cp hello.sh gs://${BUCKET}/hello.sh

由於 pig fs 指令使用 Hadoop 路徑，請將指令碼從 Cloud Storage 複製到指定為 file:/// 的目的地，確保指令碼位於本機檔案系統而非 HDFS。後續的 sh 指令會自動參照本機檔案系統，不需要 file:/// 前置字串。

gcloud dataproc jobs submit pig --cluster=${CLUSTER} --region=${REGION} \
    -e='fs -cp -f gs://${BUCKET}/hello.sh file:///tmp/hello.sh; sh chmod 750 /tmp/hello.sh; sh /tmp/hello.sh'

或者，由於 Managed Service for Apache Spark 工作提交 --jars 引數會將檔案暫存到為工作生命週期建立的暫時目錄，因此您可以將 Cloud Storage 殼層指令碼指定為 --jars 引數：

gcloud dataproc jobs submit pig --cluster=${CLUSTER} --region=${REGION} \
    --jars=gs://${BUCKET}/hello.sh \
    -e='sh chmod 750 ${PWD}/hello.sh; sh ${PWD}/hello.sh'

請注意，--jars 參數也可以引用本地腳本：

gcloud dataproc jobs submit pig --cluster=${CLUSTER} --region=${REGION} \
    --jars=hello.sh \
    -e='sh chmod 750 ${PWD}/hello.sh; sh ${PWD}/hello.sh'