Invia un job

Puoi inviare un job a un cluster Managed Service for Apache Spark esistente utilizzando una richiesta HTTP o programmatica dell'API jobs.submit, utilizzando lo strumento a riga di comando gcloud di Google Cloud CLI in una finestra del terminale locale o in Cloud Shell oppure dalla consoleGoogle Cloud aperta in un browser locale. Puoi anche eseguire l'accesso SSH all'istanza master nel cluster, quindi eseguire un job direttamente dall'istanza senza utilizzare Managed Service for Apache Spark.

Concorrenza dei job: puoi configurare il numero massimo di job Managed Service for Apache Spark simultanei con la proprietà del cluster dataproc:dataproc.scheduler.max-concurrent-jobs quando crei un cluster. Se il valore di questa proprietà non è impostato, il limite massimo di job simultanei viene calcolato come max((masterMemoryMb - 3584) / masterMemoryMbPerJob, 5). masterMemoryMb è determinato dal tipo di macchina della VM master. masterMemoryMbPerJob è 1024 per impostazione predefinita, ma è configurabile al momento della creazione del cluster con la proprietà del cluster dataproc:dataproc.scheduler.driver-size-mb.

Come inviare un job

Isolamento del classpath e JAR personalizzati: non copiare "fat JAR" personalizzati o JAR di bundle (come il runtime Apache Iceberg o i bundle Google Cloud ) direttamente nelle directory di sistema del cluster, come /usr/lib/spark/jars/. L'inserimento di JAR personalizzati nelle directory di sistema inquina il classpath dell'agente Managed Service for Apache Spark con dipendenze transitive (come le librerie Guava o Hadoop) che entrano in conflitto con le librerie integrate dell'agente. Questo conflitto può causare errori di risoluzione delle classi (come ClassNotFoundException o NoClassDefFoundError) durante l'esecuzione di operazioni di gestione dei job, come l'annullamento di un job, causando l'arresto anomalo dell'agente e lasciando le applicazioni YARN orfane.

Utilizza invece uno dei seguenti approcci supportati:

  • Dipendenze specifiche del job: specifica il percorso Cloud Storage dei tuoi JAR quando invii il job utilizzando il flag --jars in Google Cloud CLI, il campo File jar nella console Google Cloud o il campo jarFileUris nell'API. Spark distribuisce le dipendenze al driver e agli esecutori per il job senza compromettere il classpath dell'agente.
  • Dipendenze a livello di cluster: specifica le dipendenze durante la creazione del cluster utilizzando la proprietà del cluster spark:spark.jars: 
    --properties="spark:spark.jars=gs://YOUR_BUCKET/jar-1.jar,gs://YOUR_BUCKET/jar-2.jar"
    In questo modo, Managed Service for Apache Spark configura /etc/spark/conf/spark-defaults.conf, distribuendo automaticamente le dipendenze ai classpath del driver e dell'esecutore Spark per tutti i job, lasciando l'agente Managed Service for Apache Spark locale del nodo isolato in modo pulito.

Console

Apri la pagina Invia un job di Managed Service for Apache Spark nella console Google Cloud nel browser.

Esempio di job Spark

Per inviare un job Spark di esempio, compila i campi della pagina Invia un job come segue:

  1. In Seleziona il tipo di computing per il job, seleziona Spark su Compute Engine.
  2. Nella sezione Dettagli job cluster, specifica un ID job (o accetta l'ID predefinito) e seleziona il tuo Cluster dall'elenco dei cluster.
  3. Nella sezione File e artefatti:
    1. Imposta Tipo di prestazione su Spark (Scala/Java) (Spark).
    2. In Configurazione driver, seleziona Classe principale e imposta Classe principale su org.apache.spark.examples.SparkPi.
    3. Imposta Argomenti sull'argomento singolo 1000.
    4. Aggiungi file:///usr/lib/spark/examples/jars/spark-examples.jar a File jar (o al campo API jarFileUris):
      1. file:/// indica uno schema LocalFileSystem di Hadoop. Managed Service for Apache Spark ha installato /usr/lib/spark/examples/jars/spark-examples.jar sul nodo master del cluster quando ha creato il cluster. Questo percorso viene utilizzato solo per gli esempi preinstallati forniti da Managed Service for Apache Spark. Non copiare i JAR personalizzati nelle directory di sistema (consulta la sezione Attenzione in Come inviare un job).
      2. In alternativa, puoi specificare un percorso Cloud Storage (gs://your-bucket/your-jarfile.jar) o un percorso Hadoop Distributed File System (hdfs://path-to-jar.jar) per uno dei tuoi jar. Se invii il job utilizzando l'API, specifica questo percorso nel campo jarFileUris.

Fai clic su Invia per avviare il job. Una volta avviato, il job viene aggiunto all'elenco Job.

Fai clic sull'ID job per aprire la pagina Job, dove puoi visualizzare l'output del driver del job. Poiché questo job produce righe di output lunghe che superano la larghezza della finestra del browser, puoi selezionare la casella A capo automatico per visualizzare tutto il testo di output e mostrare il risultato calcolato per pi.

Puoi visualizzare l'output del driver del job dalla riga di comando utilizzando il comando gcloud dataproc jobs wait mostrato di seguito (per saperne di più, consulta Visualizza l'output del job - COMANDO GCLOUD). Copia e incolla l'ID progetto come valore per il flag --project e l'ID job (mostrato nell'elenco Job) come argomento finale.

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

Ecco alcuni snippet dell'output del driver per il job SparkPi di esempio:

...
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

Per inviare un job a un cluster Managed Service for Apache Spark, esegui il comando gcloud CLI gcloud dataproc jobs submit localmente in una finestra del terminale o in Cloud Shell. 

gcloud dataproc jobs submit job-command \
    --cluster=cluster-name \
    --region=region \
    other dataproc-flags \
    -- job-args
Esempio di invio di un job PySpark
  1. Elenca i hello-world.py accessibili pubblicamente che si trovano in Cloud Storage. 
    gcloud storage cat gs://dataproc-examples/pyspark/hello-world/hello-world.py
    Elenco dei file:

    #!/usr/bin/python
import pyspark
sc = pyspark.SparkContext()
rdd = sc.parallelize(['Hello,', 'world!'])
words = sorted(rdd.collect())
print(words)
  2. Invia il job PySpark a Managed Service for Apache Spark. 
    gcloud dataproc jobs submit pyspark \
    gs://dataproc-examples/pyspark/hello-world/hello-world.py \
    --cluster=cluster-name  \
    --region=region
    Output del terminale: 
    Waiting for job output...
…
['Hello,', 'world!']
Job finished successfully.
Esempio di invio di un job Spark
  1. Esegui l'esempio SparkPi preinstallato sul nodo master del cluster Managed Service for Apache Spark. (Il percorso file:///usr/lib/spark/examples/jars/spark-examples.jar è solo per gli esempi preinstallati; per le dipendenze JAR personalizzate, consulta l'avviso in Come inviare un job.) 
    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
    Output del terminale: 
    Job [54825071-ae28-4c5b-85a5-58fae6a597d6] submitted.
Waiting for job output…
…
Pi is roughly 3.14177148
…
Job finished successfully.
…

REST

Questa sezione mostra come inviare un job Spark per calcolare il valore approssimativo di pi utilizzando l'API jobs.submit di Managed Service for Apache Spark.

Prima di utilizzare i dati della richiesta, apporta le sostituzioni seguenti:

  • project-id: Google Cloud ID progetto
  • region: regione del cluster
  • clusterName: il nome del cluster

Metodo HTTP e URL: 

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

Corpo JSON della richiesta: 

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

Per inviare la richiesta, espandi una di queste opzioni:

Dovresti ricevere una risposta JSON simile alla seguente:

{
  "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"
}

Java

  1. Installa la libreria client

  2. Configura le Credenziali predefinite dell'applicazione

  3. Esegui il codice 
    
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. Installa la libreria client

  2. Configura le credenziali predefinite dell'applicazione

  3. Esegui il codice 
    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. Installa la libreria client

  2. Configura le credenziali predefinite dell'applicazione

  3. Esegui il codice 
    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. Installa la libreria client

  2. Configura le credenziali predefinite dell'applicazione

  3. Esegui il codice 
    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}`);

Invia un job direttamente sul tuo cluster

Se vuoi eseguire un job direttamente sul tuo cluster senza utilizzare Managed Service for Apache Spark, connettiti tramite SSH al nodo master del tuo cluster, quindi esegui il job sul nodo master.

Dopo aver stabilito una connessione SSH all'istanza master della VM, esegui i comandi in una finestra del terminale sul nodo master del cluster per:

  1. Apri una shell Spark.
  2. Esegui un job Spark per contare il numero di righe in un file Python "hello-world" (di sette righe) che si trova in un file Cloud Storage accessibile pubblicamente.

  3. Esci dalla shell.

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

Esegui job bash su Managed Service for Apache Spark

Potresti voler eseguire uno script bash come job Managed Service for Apache Spark, perché i motori che utilizzi non sono supportati come tipo di job Managed Service for Apache Spark di primo livello o perché devi eseguire una configurazione o un calcolo aggiuntivi degli argomenti prima di avviare un job utilizzando hadoop o spark-submit dallo script.

Esempio di Pig

Supponiamo che tu abbia copiato uno script bash hello.sh in Cloud Storage:

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

Poiché il comando pig fs utilizza i percorsi Hadoop, copia lo script da Cloud Storage a una destinazione specificata come file:/// per assicurarti che si trovi nel file system locale anziché in HDFS. I comandi sh successivi fanno riferimento automaticamente al file system locale e non richiedono il prefisso 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'

In alternativa, poiché l'argomento --jars di invio dei job Managed Service for Apache Spark inserisce un file in una directory temporanea creata per la durata del job, puoi specificare lo script shell di Cloud Storage come argomento --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'

Tieni presente che l'argomento --jars può anche fare riferimento a uno script locale:

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