Envie um job

É possível enviar um job para um cluster do Serviço Gerenciado para Apache Spark usando uma solicitação programática ou HTTP da API jobs.submit, usando a ferramenta de linha de comando gcloud da Google Cloud CLI em uma janela do terminal local ou no Cloud Shell, ou no Google Cloud console aberto em um navegador local. Também é possível fazer SSH na instância mestre no cluster e executar um job diretamente da instância sem usar o Serviço Gerenciado para Apache Spark.

Simultaneidade de jobs: é possível configurar o número máximo de jobs simultâneos do Serviço Gerenciado para Apache Spark com a propriedade de cluster dataproc:dataproc.scheduler.max-concurrent-jobs ao criar um cluster. Se esse valor de propriedade não for definido, o limite máximo de jobs simultâneos será calculado como max((masterMemoryMb - 3584) / masterMemoryMbPerJob, 5). masterMemoryMb é determinado pelo tipo de máquina da VM mestre. masterMemoryMbPerJob é 1024 por padrão, mas é configurável na criação do cluster com a propriedade de cluster dataproc:dataproc.scheduler.driver-size-mb.

Como enviar um job

Isolamento de classpath e JARs personalizados: não copie "fat JARs" personalizados ou pacotes JAR (como o ambiente de execução do Apache Iceberg ou pacotes do Google Cloud ) diretamente para diretórios do sistema do cluster, como /usr/lib/spark/jars/. Colocar JARs personalizados em diretórios do sistema polui o caminho de classe do agente do Serviço Gerenciado para Apache Spark com dependências transitivas (como bibliotecas Guava ou Hadoop) que entram em conflito com as bibliotecas integradas do agente. Esse conflito pode causar erros de resolução de classe (como ClassNotFoundException ou NoClassDefFoundError) ao realizar operações de gerenciamento de jobs, como cancelar um job, fazendo com que o agente falhe e deixando os aplicativos do YARN órfãos.

Em vez disso, use uma das seguintes abordagens compatíveis:

  • Dependências específicas do job: especifique o caminho do Cloud Storage para os JARs ao enviar o job usando a flag --jars na CLI do Google Cloud, o campo Arquivos JAR no console do Google Cloud ou o campo jarFileUris na API. O Spark distribui as dependências para o driver e os executores do job sem poluir o caminho de classe do agente.
  • Dependências em todo o cluster: especifique as dependências durante a criação do cluster usando a propriedade de cluster spark:spark.jars: 
    --properties="spark:spark.jars=gs://YOUR_BUCKET/jar-1.jar,gs://YOUR_BUCKET/jar-2.jar"
    Isso instrui o Serviço Gerenciado para Apache Spark a configurar /etc/spark/conf/spark-defaults.conf, distribuindo automaticamente as dependências para os caminhos de classe do driver e do executor do Spark para todos os jobs, mantendo o agente do Serviço Gerenciado para Apache Spark local do nó isolado.

Console

Abra a página Enviar um job do Serviço Gerenciado para Apache Spark no console do Google Cloud no navegador.

Exemplo de job do Spark

Para enviar um job de amostra do Spark, preencha os campos na página Enviar um job da seguinte maneira:

  1. Em Selecionar o tipo de computação para o job, selecione Spark no Compute Engine.
  2. Na seção Detalhes do job do cluster, especifique um código da tarefa (ou aceite o ID padrão) e selecione o Cluster na lista de clusters.
  3. Na seção Arquivos e artefatos:
    1. Defina Tipo de serviço como Spark (Scala / Java) (Spark).
    2. Em Configuração do driver, selecione Classe principal e defina Classe principal como org.apache.spark.examples.SparkPi.
    3. Defina Argumentos como o argumento único 1000.
    4. Adicione file:///usr/lib/spark/examples/jars/spark-examples.jar aos arquivos JAR (ou ao campo jarFileUris da API):
      1. file:/// indica um esquema de LocalFileSystem do Hadoop. O Serviço Gerenciado para Apache Spark instalou o /usr/lib/spark/examples/jars/spark-examples.jar no nó mestre do cluster quando o criou. Esse caminho é usado apenas para exemplos pré-instalados fornecidos pelo Serviço Gerenciado para Apache Spark. Não copie JARs personalizados para diretórios do sistema (consulte o aviso em Como enviar um job).
      2. Como alternativa, é possível especificar um caminho do Cloud Storage (gs://your-bucket/your-jarfile.jar) ou um caminho do sistema de arquivos distribuídos do Hadoop (hdfs://path-to-jar.jar) para um dos seus jars. Se você enviar o job usando a API, especifique esse caminho no campo jarFileUris.

Clique em Enviar para iniciar o job. Depois de iniciado, o job será adicionado à lista.

Clique no código da tarefa para abrir a página Jobs, em que é possível ver a saída do driver da tarefa. Como esse job produz linhas de saída longas que excedem a largura da janela do navegador, marque a caixa Quebra de linha para exibir todo o texto de saída e mostrar o resultado calculado para pi.

É possível ver a saída do driver do job na linha de comando usando o comando gcloud dataproc jobs wait mostrado abaixo. Para mais informações, consulte Ver saída do job: COMANDO GCLOUD. Copie e cole o ID do projeto como o valor da flag --project e o ID do job (mostrado na lista de jobs) como o argumento final.

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

Veja snippets da saída do driver para o job de amostra 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 um job a um cluster do Serviço Gerenciado para Apache Spark, execute o comando gcloud dataproc jobs submit da CLI gcloud localmente em uma janela do terminal ou no Cloud Shell. 

gcloud dataproc jobs submit job-command \
    --cluster=cluster-name \
    --region=region \
    other dataproc-flags \
    -- job-args
Exemplo de envio de job do PySpark
  1. Liste os hello-world.py acessíveis publicamente localizados no Cloud Storage. 
    gcloud storage cat gs://dataproc-examples/pyspark/hello-world/hello-world.py
    Lista de arquivos:

    #!/usr/bin/python
import pyspark
sc = pyspark.SparkContext()
rdd = sc.parallelize(['Hello,', 'world!'])
words = sorted(rdd.collect())
print(words)
  2. Envie o job do PySpark para o Serviço Gerenciado para Apache Spark. 
    gcloud dataproc jobs submit pyspark \
    gs://dataproc-examples/pyspark/hello-world/hello-world.py \
    --cluster=cluster-name  \
    --region=region
    Saída do terminal: 
    Waiting for job output...
…
['Hello,', 'world!']
Job finished successfully.
Exemplo de envio de job do Spark
  1. Execute o exemplo do SparkPi pré-instalado no nó mestre do cluster do Serviço Gerenciado para Apache Spark. O caminho file:///usr/lib/spark/examples/jars/spark-examples.jar é apenas para exemplos pré-instalados. Para dependências JAR personalizadas, consulte o aviso em Como enviar um 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
    Saída do terminal: 
    Job [54825071-ae28-4c5b-85a5-58fae6a597d6] submitted.
Waiting for job output…
…
Pi is roughly 3.14177148
…
Job finished successfully.
…

REST

Nesta seção, mostramos como enviar um job do Spark para calcular o valor aproximado de pi usando a API jobs.submit do Serviço Gerenciado para Apache Spark.

Antes de usar os dados da solicitação abaixo, faça as substituições a seguir:

  • project-id: Google Cloud ID do projeto
  • region: região do cluster
  • clusterName: nome do cluster

Método HTTP e URL:

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

Corpo JSON da solicitação:

{
  "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 a solicitação, expanda uma destas opções:

Você receberá uma resposta JSON semelhante a esta:

{
  "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. Instalar a biblioteca de cliente

  2. Configurar as credenciais padrão do aplicativo

  3. Executar o 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. Instalar a biblioteca de cliente

  2. Configurar as credenciais padrão do aplicativo

  3. Executar o 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. Instalar a biblioteca de cliente

  2. Configurar as credenciais padrão do aplicativo

  3. Executar o 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. Instalar a biblioteca de cliente

  2. Configurar as credenciais padrão do aplicativo

  3. Executar o 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}`);

Enviar um trabalho diretamente no cluster

Se você quiser executar um job diretamente no cluster sem usar o Serviço Gerenciado para Apache Spark, faça login por SSH no nó mestre do cluster e execute o job no nó mestre.

Depois de estabelecer uma conexão SSH com a instância mestre da VM, execute comandos em uma janela de terminal no nó mestre do cluster para:

  1. abrir um shell do Spark;
  2. Execute um job do Spark para contar o número de linhas em um arquivo Python "hello-world" (de sete linhas) localizado em um arquivo do Cloud Storage acessível publicamente.

  3. sair do 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

Executar jobs do Bash no Serviço Gerenciado para Apache Spark

Talvez você queira executar um script bash como seu job do Serviço Gerenciado para Apache Spark, seja porque os mecanismos usados não são compatíveis como um tipo de serviço de nível superior do Serviço Gerenciado para Apache Spark ou porque você precisa fazer uma configuração ou um cálculo adicional de argumentos antes de iniciar um job usando hadoop ou spark-submit do seu script.

Exemplo de Python

Suponha que você tenha copiado um script hello.sh bash no Cloud Storage:

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

Como o comando pig fs usa caminhos do Hadoop, copie o script do Cloud Storage para um destino especificado como file:/// para garantir que ele esteja no sistema de arquivos local em vez do HDFS. Os comandos sh subsequentes fazem referência ao sistema de arquivos local automaticamente e não exigem o prefixo 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, como o argumento --jars de envio de jobs do Serviço Gerenciado para Apache Spark prepara um arquivo em um diretório temporário criado para o ciclo de vida do job, é possível especificar o script de shell do Cloud Storage como um 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'

Observe que o argumento --jars também pode fazer referência a um 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'