Envie um trabalho

Pode enviar uma tarefa para um cluster do Dataproc existente através de um pedido HTTP ou programático jobs.submit da API Dataproc, usando a ferramenta de linha de comandos gcloud da CLI do Google Cloud numa janela de terminal local ou no Cloud Shell, ou a partir da Google Cloud consola aberta num navegador local. Também pode usar SSH na instância principal no cluster e, em seguida, executar uma tarefa diretamente a partir da instância sem usar o serviço Dataproc.

Como enviar um trabalho

Consola

Abra a página Dataproc Enviar uma tarefa na Google Cloud consola no seu navegador.

Exemplo de tarefa do Spark

Para enviar um exemplo de tarefa do Spark, preencha os campos na página Enviar uma tarefa, da seguinte forma:

  1. Selecione o nome do cluster na lista de clusters.
  2. Defina o Tipo de serviço como Spark.
  3. Defina Main class or jar como org.apache.spark.examples.SparkPi.
  4. Defina Arguments para o argumento único 1000.
  5. Adicione file:///usr/lib/spark/examples/jars/spark-examples.jar a ficheiros JAR:
    1. file:/// denota um esquema Hadoop LocalFileSystem. O Dataproc instalou o /usr/lib/spark/examples/jars/spark-examples.jar no nó principal do cluster quando criou o cluster.
    2. Em alternativa, pode especificar um caminho do Cloud Storage (gs://your-bucket/your-jarfile.jar) ou um caminho do sistema de ficheiros distribuídos do Hadoop (hdfs://path-to-jar.jar) para um dos seus ficheiros JAR.

Clique em Enviar para iniciar a tarefa. Assim que a tarefa for iniciada, é adicionada à lista de tarefas.

Clique no ID do trabalho para abrir a página Trabalhos, onde pode ver o resultado do controlador do trabalho. Uma vez que esta tarefa produz linhas de saída longas que excedem a largura da janela do navegador, pode selecionar a caixa Quebra de linha para apresentar todo o texto de saída dentro da vista, de modo a apresentar o resultado calculado para pi.

Pode ver o resultado do controlador da tarefa a partir da linha de comandos através do comando gcloud dataproc jobs wait apresentado abaixo (para mais informações, consulte Ver resultado da tarefa – COMANDO GCLOUD). Copie e cole o ID do projeto como o valor da flag --project e o ID do trabalho (apresentado na lista de trabalhos) como o argumento final.

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

Seguem-se fragmentos do resultado do controlador para o exemplo de SparkPi tarefa enviado acima:

...
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 uma tarefa para um cluster do Dataproc, execute o comando gcloud dataproc jobs submit localmente numa janela de 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 tarefas do PySpark
  1. Apresente uma lista dos hello-world.py acessíveis publicamente localizados no Cloud Storage. 
    gcloud storage cat gs://dataproc-examples/pyspark/hello-world/hello-world.py
    Ficheiro de listagem:

    #!/usr/bin/python
import pyspark
sc = pyspark.SparkContext()
rdd = sc.parallelize(['Hello,', 'world!'])
words = sorted(rdd.collect())
print(words)
  2. Envie a tarefa Pyspark para o Dataproc. 
    gcloud dataproc jobs submit pyspark \
    gs://dataproc-examples/pyspark/hello-world/hello-world.py \
    --cluster=cluster-name  \
    --region=region
    Resultado do terminal: 
    Waiting for job output...
…
['Hello,', 'world!']
Job finished successfully.
Exemplo de envio de tarefa do Spark
  1. Execute o exemplo SparkPi pré-instalado no nó principal do cluster do Dataproc. 
    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 do terminal: 
    Job [54825071-ae28-4c5b-85a5-58fae6a597d6] submitted.
Waiting for job output…
…
Pi is roughly 3.14177148
…
Job finished successfully.
…

REST

Esta secção mostra como enviar uma tarefa do Spark para calcular o valor aproximado de pi através da API jobs.submit do Dataproc.

Antes de usar qualquer um dos dados do pedido, faça as seguintes substituições:

  • 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 do pedido: 

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

Deve receber uma resposta JSON semelhante à seguinte:

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

  2. Configure as credenciais padrão da aplicação

  3. Execute 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. Instale a biblioteca cliente

  2. Configure as credenciais padrão da aplicação

  3. Execute 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. Instale a biblioteca cliente

  2. Configure as credenciais padrão da aplicação

  3. Execute 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. Instale a biblioteca cliente

  2. Configure as credenciais padrão da aplicação

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

Envie um trabalho diretamente no seu cluster

Se quiser executar uma tarefa diretamente no cluster sem usar o serviço Dataproc, inicie sessão no nó principal do cluster através de SSH e, em seguida, execute a tarefa no nó principal.

Depois de estabelecer uma ligação SSH à instância principal da VM, execute comandos numa janela de terminal no nó principal do cluster para:

  1. Abra um shell do Spark.
  2. Execute uma tarefa simples do Spark para contar o número de linhas num ficheiro "hello-world" do Python (com sete linhas) localizado num ficheiro do Cloud Storage acessível publicamente.

  3. Saia da 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

Execute tarefas bash no Dataproc

Pode querer executar um script bash como tarefa do Dataproc, porque os motores que usa não são suportados como um tipo de tarefa do Dataproc de nível superior ou porque precisa de fazer uma configuração ou um cálculo adicional de argumentos antes de iniciar uma tarefa com hadoop ou spark-submit a partir do seu script.

Exemplo de porco

Suponha que copiou um script bash hello.sh para o Cloud Storage:

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

Uma vez que o comando pig fs usa caminhos do Hadoop, copie o script do armazenamento na nuvem para um destino especificado como file:/// para se certificar de que está no sistema de ficheiros local em vez do HDFS. Os comandos shsubsequentes fazem referência automaticamente ao sistema de ficheiros local e não requerem 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'

Em alternativa, uma vez que as tarefas do Dataproc enviam argumentos --jars para um ficheiro num diretório temporário criado durante a duração da tarefa, pode especificar o seu 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'

Tenha em atenção que o argumento --jars também pode referenciar 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'