Envoyer une tâche

Vous pouvez envoyer un job à un cluster Managed Service pour Apache Spark existant à l'aide d'une requête HTTP ou programmatique de l'API jobs.submit, à l'aide de l'outil de ligne de commande gcloud de Google Cloud CLI dans une fenêtre de terminal locale ou dans Cloud Shell, ou depuis la consoleGoogle Cloud ouverte dans un navigateur local. Vous pouvez également vous connecter en SSH à l'instance maître de votre cluster, puis exécuter un job directement à partir de l'instance sans utiliser Managed Service pour Apache Spark.

Simultanéité des jobs : vous pouvez configurer le nombre maximal de jobs Managed Service pour Apache Spark simultanés à l'aide de la propriété de cluster dataproc:dataproc.scheduler.max-concurrent-jobs lorsque vous créez un cluster. Si cette valeur de propriété n'est pas définie, la limite supérieure du nombre de jobs simultanés est calculée comme suit : max((masterMemoryMb - 3584) / masterMemoryMbPerJob, 5). masterMemoryMb est déterminé par le type de machine de la VM maître. masterMemoryMbPerJob est défini sur 1024 par défaut, mais peut être configuré lors de la création du cluster avec la propriété de cluster dataproc:dataproc.scheduler.driver-size-mb.

Envoyer un job

Isolation du chemin de classe et fichiers JAR personnalisés : ne copiez pas de fichiers JAR personnalisés volumineux ni de bundles JAR (comme l'environnement d'exécution Apache Iceberg ou les bundles Google Cloud ) directement dans les répertoires système du cluster, tels que /usr/lib/spark/jars/. Le fait de placer des fichiers JAR personnalisés dans des répertoires système pollue le chemin de classe de l'agent Managed Service pour Apache Spark avec des dépendances transitives (telles que les bibliothèques Guava ou Hadoop) qui entrent en conflit avec les bibliothèques intégrées de l'agent. Ce conflit peut entraîner des erreurs de résolution de classe (telles que ClassNotFoundException ou NoClassDefFoundError) lors de l'exécution d'opérations de gestion des jobs, comme l'annulation d'un job, ce qui entraîne le plantage de l'agent et laisse les applications YARN orphelines.

Utilisez plutôt l'une des approches compatibles suivantes :

  • Dépendances spécifiques au job : spécifiez le chemin d'accès Cloud Storage à vos fichiers JAR lorsque vous envoyez le job à l'aide de l'option --jars dans la CLI Google Cloud CLI, du champ Fichiers JAR dans la console Google Cloud ou du champ jarFileUris dans l'API. Spark distribue les dépendances au pilote et aux exécuteurs du job sans polluer le chemin de classe de l'agent.
  • Dépendances à l'échelle du cluster : spécifiez les dépendances lors de la création du cluster à l'aide de la propriété de cluster spark:spark.jars : 
    --properties="spark:spark.jars=gs://YOUR_BUCKET/jar-1.jar,gs://YOUR_BUCKET/jar-2.jar"
    Cela indique à Managed Service pour Apache Spark de configurer /etc/spark/conf/spark-defaults.conf, en distribuant automatiquement les dépendances aux chemins de classe du pilote et de l'exécuteur Spark pour tous les jobs, tout en laissant l'agent Managed Service pour Apache Spark local au nœud isolé.

Console

Ouvrez la page Envoyer un job de Managed Service pour Apache Spark dans la console Google Cloud de votre navigateur.

Exemple de tâche Spark

Pour envoyer un exemple de job Spark, remplissez les champs de la page Envoyer un job comme suit :

  1. Sous Sélectionner le type de calcul pour le job, sélectionnez Spark sur Compute Engine.
  2. Dans la section Détails du job de cluster, spécifiez un ID de job (ou acceptez l'ID par défaut) et sélectionnez votre cluster dans la liste des clusters.
  3. Dans la section Fichiers et artefacts :
    1. Définissez le type de job sur Spark (Scala/Java) (Spark).
    2. Sous Configuration du pilote, sélectionnez Classe principale, puis définissez Classe principale sur org.apache.spark.examples.SparkPi.
    3. Définissez le champ Arguments sur l'argument unique 1000.
    4. Ajoutez file:///usr/lib/spark/examples/jars/spark-examples.jar à Fichiers JAR (ou au champ d'API jarFileUris) :
      1. file:/// désigne un schéma LocalFileSystem Hadoop. Managed Service pour Apache Spark a installé /usr/lib/spark/examples/jars/spark-examples.jar sur le nœud maître du cluster lors de sa création. Ce chemin d'accès n'est utilisé que pour les exemples préinstallés fournis par Managed Service pour Apache Spark. Ne copiez pas de fichiers JAR personnalisés dans les répertoires système (consultez la mise en garde dans Envoyer un job).
      2. Vous pouvez également spécifier un chemin d'accès Cloud Storage (gs://your-bucket/your-jarfile.jar) ou un chemin d'accès au système de fichiers distribué Hadoop (hdfs://path-to-jar.jar) vers l'un de vos fichiers JAR. (Si vous envoyez le job à l'aide de l'API, spécifiez ce chemin d'accès dans le champ jarFileUris.)

Cliquez sur Envoyer pour démarrer la tâche. Une fois la tâche démarrée, elle est ajoutée à la liste des Tâches.

Cliquez sur l'ID de job pour ouvrir la page Jobs, où vous pouvez afficher la sortie du pilote du job. Comme ce job produit de longues lignes de sortie qui dépassent la largeur de la fenêtre du navigateur, vous pouvez cocher la case Renvoi à la ligne automatique pour afficher l'intégralité du texte de sortie afin d'afficher le résultat calculé pour pi.

Vous pouvez afficher la sortie du pilote de votre job à partir de la ligne de commande à l'aide de la commande gcloud dataproc jobs wait présentée ci-dessous (pour en savoir plus, consultez Afficher la sortie du job – COMMANDE GCLOUD). Copiez et collez l'ID de votre projet comme valeur de l'option --project et l'ID de votre job (affiché dans la liste des jobs) comme argument final.

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

Voici des extraits de la sortie du pilote pour l'exemple de job 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

Pour envoyer un job à un cluster Managed Service pour Apache Spark, exécutez la commande gcloud CLI gcloud dataproc jobs submit localement dans une fenêtre de terminal ou dans Cloud Shell. 

gcloud dataproc jobs submit job-command \
    --cluster=cluster-name \
    --region=region \
    other dataproc-flags \
    -- job-args
Exemple d'envoi de job PySpark
  1. Répertoriez l'élément accessible au public hello-world.py, situé dans Cloud Storage. 
    gcloud storage cat gs://dataproc-examples/pyspark/hello-world/hello-world.py
    Liste des fichiers :

    #!/usr/bin/python
import pyspark
sc = pyspark.SparkContext()
rdd = sc.parallelize(['Hello,', 'world!'])
words = sorted(rdd.collect())
print(words)
  2. Envoyez le job PySpark à Managed Service pour Apache Spark. 
    gcloud dataproc jobs submit pyspark \
    gs://dataproc-examples/pyspark/hello-world/hello-world.py \
    --cluster=cluster-name  \
    --region=region
    Résultat du terminal : 
    Waiting for job output...
…
['Hello,', 'world!']
Job finished successfully.
Exemple d'envoi de tâche Spark
  1. Exécutez l'exemple SparkPi préinstallé sur le nœud maître du cluster Managed Service pour Apache Spark. (Le chemin d'accès file:///usr/lib/spark/examples/jars/spark-examples.jar ne concerne que les exemples préinstallés. Pour les dépendances JAR personnalisées, consultez la mise en garde dans la section Envoyer 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
    Résultat du terminal : 
    Job [54825071-ae28-4c5b-85a5-58fae6a597d6] submitted.
Waiting for job output…
…
Pi is roughly 3.14177148
…
Job finished successfully.
…

REST

Cette section explique comment envoyer un job Spark pour calculer la valeur approximative de pi à l'aide de l'API jobs.submit de Managed Service pour Apache Spark.

Avant d'utiliser les données de requête, effectuez les remplacements suivants :

  • project-id : Google Cloud ID du projet
  • region : région du cluster
  • clusterName : nom du cluster

Méthode HTTP et URL :

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

Corps JSON de la requête :

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

Pour envoyer votre requête, développez l'une des options suivantes :

Vous devriez recevoir une réponse JSON de ce type :

{
  "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. Installer la bibliothèque cliente

  2. Configurer les identifiants par défaut de l'application

  3. Exécuter le code 
    
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. Installer la bibliothèque cliente

  2. Configurer les identifiants par défaut de l'application

  3. Exécutez le code. 
    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. Installer la bibliothèque cliente

  2. Configurer les identifiants par défaut de l'application

  3. Exécuter le code 
    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. Installer la bibliothèque cliente

  2. Configurer les identifiants par défaut de l'application

  3. Exécutez le code. 
    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}`);

Envoyer une tâche directement sur votre cluster

Si vous souhaitez exécuter un job directement sur votre cluster sans utiliser Managed Service pour Apache Spark, connectez-vous en SSH au nœud maître de votre cluster, puis exécutez le job sur le nœud maître.

Après avoir établi une connexion SSH à l'instance maître de VM, exécutez des commandes dans une fenêtre de terminal sur le nœud maître du cluster pour :

  1. Ouvrir une interface système Spark.
  2. Exécutez un job Spark pour compter le nombre de lignes d'un fichier Python "hello-world" (de sept lignes) situé dans un fichier Cloud Storage accessible publiquement.

  3. Fermer l'interface système.

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

Exécuter des jobs Bash sur Managed Service pour Apache Spark

Vous pouvez exécuter un script bash en tant que job Managed Service pour Apache Spark, soit parce que les moteurs que vous utilisez ne sont pas compatibles en tant que type de job Managed Service pour Apache Spark de premier niveau, soit parce que vous devez effectuer une configuration ou un calcul d'arguments supplémentaires avant de lancer un job à l'aide de hadoop ou spark-submit à partir de votre script.

Exemple d'instruction Pig

Supposons que vous ayez copié un script bash "hello.sh" dans Cloud Storage :

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

Comme la commande pig fs utilise des chemins Hadoop, copiez le script de Cloud Storage vers une destination spécifiée en tant que file:/// pour vous assurer qu'il se trouve sur le système de fichiers local plutôt que sur HDFS. Les commandes sh suivantes font automatiquement référence au système de fichiers local et ne nécessitent pas le préfixe 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'

Sinon, étant donné que l'argument --jars d'envoi de jobs Managed Service pour Apache Spark préproduit un fichier dans un répertoire temporaire créé pour la durée de vie du job, vous pouvez spécifier votre script shell Cloud Storage en tant qu'argument --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'

Notez que l'argument --jars peut également faire référence à un script local :

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