Envía un trabajo

Puedes enviar un trabajo a un clúster existente de Managed Service para Apache Spark con una solicitud programática o HTTP de la API jobs.submit, con la herramienta de línea de comandos gcloud de Google Cloud CLI en una ventana de terminal local o en Cloud Shell, o desde la consola deGoogle Cloud abierta en un navegador local. También puedes establecer una conexión SSH con la instancia principal de tu clúster y, luego, ejecutar un trabajo directamente desde la instancia sin usar Managed Service para Apache Spark.

Simultaneidad de trabajos: Puedes configurar la cantidad máxima de trabajos simultáneos de Managed Service para Apache Spark con la propiedad de clúster dataproc:dataproc.scheduler.max-concurrent-jobs cuando creas un clúster. Si no se establece este valor de propiedad, el límite superior de trabajos simultáneos se calcula como max((masterMemoryMb - 3584) / masterMemoryMbPerJob, 5). masterMemoryMb se determina según el tipo de máquina de la VM principal. masterMemoryMbPerJob es 1024 de forma predeterminada, pero se puede configurar en la creación del clúster con la propiedad de clúster dataproc:dataproc.scheduler.driver-size-mb.

Cómo enviar un trabajo

Aislamiento de ruta de clase y JAR personalizados: No copies "JAR pesados" personalizados ni paquetes JAR (como el entorno de ejecución de Apache Iceberg o los paquetes de Google Cloud ) directamente en los directorios del sistema del clúster, como /usr/lib/spark/jars/. Colocar archivos JAR personalizados en directorios del sistema contamina la ruta de clase del agente de Managed Service para Apache Spark con dependencias transitivas (como las bibliotecas de Guava o Hadoop) que entran en conflicto con las bibliotecas integradas del agente. Este conflicto puede causar errores de resolución de clases (como ClassNotFoundException o NoClassDefFoundError) cuando se realizan operaciones de administración de trabajos, como cancelar un trabajo, lo que provoca que el agente falle y deje huérfanas las aplicaciones de YARN.

En su lugar, usa uno de los siguientes enfoques admitidos:

  • Dependencias específicas del trabajo: Especifica la ruta de acceso de Cloud Storage a tus archivos JAR cuando envíes el trabajo con la marca --jars en Google Cloud CLI, el campo Archivos JAR en la consola de Google Cloud o el campo jarFileUris en la API. Spark distribuye las dependencias al controlador y a los ejecutores del trabajo sin contaminar la ruta de clase del agente.
  • Dependencias de todo el clúster: Especifica las dependencias durante la creación del clúster con la propiedad del clúster spark:spark.jars: 
    --properties="spark:spark.jars=gs://YOUR_BUCKET/jar-1.jar,gs://YOUR_BUCKET/jar-2.jar"
    Esto le indica a Managed Service para Apache Spark que configure /etc/spark/conf/spark-defaults.conf, lo que distribuye automáticamente las dependencias a las rutas de clase del controlador y del ejecutor de Spark para todos los trabajos, mientras que deja el agente de Managed Service para Apache Spark local del nodo aislado de forma limpia.

Console

Abre la página Enviar un trabajo de Managed Service para Apache Spark en la consola de Google Cloud en tu navegador.

Ejemplo de un trabajo de Spark

Para enviar un trabajo de Spark de muestra, completa los campos de la página Enviar un trabajo de la siguiente manera:

  1. En Selecciona el tipo de procesamiento para el trabajo, elige Spark en Compute Engine.
  2. En la sección Detalles del trabajo del clúster, especifica un ID de trabajo (o acepta el ID predeterminado) y selecciona tu Clúster de la lista de clústeres.
  3. En la sección Archivos y artefactos, haz lo siguiente:
    1. Establece el Tipo de trabajo en Spark (Scala/Java) (Spark).
    2. En Configuración del controlador, selecciona Clase principal y configúrala como org.apache.spark.examples.SparkPi.
    3. Establece Argumentos (Arguments) en un argumento único 1000.
    4. Agrega file:///usr/lib/spark/examples/jars/spark-examples.jar a Archivos JAR (o al campo jarFileUris de la API):
      1. file:/// denota un esquema LocalFileSystem de Hadoop. Managed Service para Apache Spark instaló /usr/lib/spark/examples/jars/spark-examples.jar en el nodo principal del clúster cuando lo creó. Esta ruta de acceso solo se usa para los ejemplos preinstalados que proporciona Managed Service para Apache Spark. No copies archivos JAR personalizados en directorios del sistema (consulta la advertencia en Cómo enviar un trabajo).
      2. Como alternativa, puedes especificar una ruta de acceso de Cloud Storage (gs://your-bucket/your-jarfile.jar) o una ruta de acceso del sistema de archivos distribuido de Hadoop (hdfs://path-to-jar.jar) a uno de tus archivos JAR. (Si envías el trabajo con la API, especifica esta ruta en el campo jarFileUris).

Haz clic en Submit (Enviar) para iniciar el trabajo. Una vez que se inicia el trabajo, se agrega a la lista Jobs (Trabajos).

Haz clic en el ID del trabajo para abrir la página Trabajos, en la que puedes ver el resultado del controlador del trabajo. Debido a que este trabajo produce líneas de salida largas que superan el ancho de la ventana del navegador, puedes marcar la casilla Ajuste de línea para que todo el texto de salida esté a la vista y se muestre el resultado calculado para pi.

Puedes ver el resultado del controlador de tu trabajo desde la línea de comandos con el comando gcloud dataproc jobs wait que se muestra a continuación (para obtener más información, consulta Visualiza el resultado del trabajo: COMANDO DE GCLOUD). Copia y pega tu ID del proyecto como el valor de la marca --project y tu ID de trabajo (que aparece en la lista de trabajos) como argumento final.

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

Estos son fragmentos de la salida del controlador para el trabajo de muestra 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

Para enviar un trabajo a un clúster de Managed Service para Apache Spark, ejecuta el comando gcloud dataproc jobs submit de gcloud CLI de forma local en una ventana de terminal o en Cloud Shell. 

gcloud dataproc jobs submit job-command \
    --cluster=cluster-name \
    --region=region \
    other dataproc-flags \
    -- job-args
Ejemplo de envío de un trabajo de PySpark
  1. Muestra el archivo hello-world.py de acceso público ubicado en Cloud Storage. 
    gcloud storage cat gs://dataproc-examples/pyspark/hello-world/hello-world.py
    Lista de archivos:

    #!/usr/bin/python
import pyspark
sc = pyspark.SparkContext()
rdd = sc.parallelize(['Hello,', 'world!'])
words = sorted(rdd.collect())
print(words)
  2. Envía el trabajo de PySpark a Managed Service para Apache Spark. 
    gcloud dataproc jobs submit pyspark \
    gs://dataproc-examples/pyspark/hello-world/hello-world.py \
    --cluster=cluster-name  \
    --region=region
    Resultado de la terminal: 
    Waiting for job output...
…
['Hello,', 'world!']
Job finished successfully.
Ejemplo de envío de un trabajo de Spark
  1. Ejecuta el ejemplo de SparkPi preinstalado en el nodo principal del clúster de Managed Service para Apache Spark. (La ruta de acceso file:///usr/lib/spark/examples/jars/spark-examples.jar es solo para ejemplos preinstalados; para las dependencias de JAR personalizadas, consulta la precaución en Cómo enviar un trabajo). 
    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
    Resultado de la terminal: 
    Job [54825071-ae28-4c5b-85a5-58fae6a597d6] submitted.
Waiting for job output…
…
Pi is roughly 3.14177148
…
Job finished successfully.
…

REST

En esta sección, se muestra cómo enviar un trabajo de Spark para calcular el valor aproximado de pi con la API de jobs.submit de Managed Service para Apache Spark.

Antes de usar cualquiera de los datos de solicitud a continuación, realiza los siguientes reemplazos:

  • project-id: ID del proyecto de Google Cloud
  • region: región del clúster
  • clusterName: nombre del clúster

Método HTTP y URL:

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

Cuerpo JSON de la solicitud:

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

Para enviar tu solicitud, expande una de estas opciones:

Deberías recibir una respuesta JSON similar a la que se muestra a continuación:

{
  "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. Instala la biblioteca cliente

  2. Configura credenciales predeterminadas de la aplicación

  3. Ejecuta el código 
    
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. Instala la biblioteca cliente

  2. Configura credenciales predeterminadas de la aplicación

  3. Ejecuta el código 
    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. Instala la biblioteca cliente

  2. Configura credenciales predeterminadas de la aplicación

  3. Ejecuta el código 
    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. Instala la biblioteca cliente

  2. Configura credenciales predeterminadas de la aplicación

  3. Ejecuta el código 
    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}`);

Envía un trabajo directamente en tu clúster

Si deseas ejecutar un trabajo directamente en tu clúster sin usar el Managed Service para Apache Spark, establece una conexión SSH con el nodo principal de tu clúster y, luego, ejecuta el trabajo en el nodo principal.

Después de establecer una conexión SSH a la instancia principal de la VM, ejecuta comandos en una ventana de terminal en el nodo principal del clúster para hacer lo siguiente:

  1. Abrir una shell de Spark
  2. Ejecuta un trabajo de Spark para contar la cantidad de líneas en un archivo de Python "hello-world" (de siete líneas) ubicado en un archivo de Cloud Storage de acceso público.

  3. Salir de la 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

Ejecuta trabajos de Bash en Managed Service para Apache Spark

Es posible que quieras ejecutar una secuencia de comandos de Bash como tu trabajo de Managed Service para Apache Spark, ya sea porque los motores que usas no se admiten como un tipo de trabajo de Managed Service para Apache Spark de nivel superior o porque necesitas realizar una configuración o un cálculo de argumentos adicionales antes de iniciar un trabajo con hadoop o spark-submit desde tu secuencia de comandos.

Ejemplo de pig

Supongamos que copiaste una secuencia de comandos bash hello.sh en Cloud Storage:

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

Dado que el comando pig fs usa rutas de Hadoop, copia la secuencia de comandos de Cloud Storage en un destino especificado como file:/// para asegurarte de que esté en el sistema de archivos local, en lugar de HDFS. Los comandos sh posteriores hacen referencia al sistema de archivos local de forma automática y no requieren el prefijo 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'

Como alternativa, dado que el argumento --jars de envío de trabajos de Managed Service para Apache Spark almacena en etapa intermedia un archivo en un directorio temporal creado para la duración del trabajo, puedes especificar tu secuencia de comandos de shell de Cloud Storage como un argumento --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'

Ten en cuenta que el argumento --jars también puede hacer referencia a una secuencia de comandos local:

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