Use o conetor do Cloud Storage com o Apache Spark

Este tutorial mostra como executar um código de exemplo que usa o conetor do Cloud Storage com o Apache Spark.

Objetivos

Escreva um trabalho simples de contagem de palavras do Spark em Java, Scala ou Python e, em seguida, execute o trabalho num cluster do Dataproc.

Todas as versões de imagens de clusters do Dataproc têm os componentes do Spark necessários para este tutorial já instalados.

Custos

Neste documento, usa os seguintes componentes faturáveis do Google Cloud:

• Compute Engine
• Dataproc
• Cloud Storage

Para gerar uma estimativa de custos com base na sua utilização projetada, use a calculadora de preços.

Os novos Google Cloud utilizadores podem ser elegíveis para uma avaliação gratuita.

Antes de começar

Execute os passos abaixo para se preparar para executar o código neste tutorial.

1. Configure o seu projeto. Se necessário, configure um projeto com as APIs Dataproc, Compute Engine e Cloud Storage ativadas, e a CLI Google Cloud instalada no seu computador local.

   Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.

   In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Roles required to select or create a project

   • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
   • Create a project: To create a project, you need the Project Creator (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

   Verify that billing is enabled for your Google Cloud project.

   Enable the Dataproc, Compute Engine, and Cloud Storage APIs.

   Roles required to enable APIs

   To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

   Enable the APIs

   Create a service account:

   1. Ensure that you have the Create Service Accounts IAM role (roles/iam.serviceAccountCreator). Learn how to grant roles.

   2. In the Google Cloud console, go to the Create service account page.

      Go to Create service account
   3. Select your project.

   4. In the Service account name field, enter a name. The Google Cloud console fills in the Service account ID field based on this name.

      In the Service account description field, enter a description. For example, Service account for quickstart.

   5. Click Create and continue.

   6. Grant the Project > Owner role to the service account.

      To grant the role, find the Select a role list, then select Project > Owner.

      Note: The Role field affects which resources the service account can access in your project. You can revoke these roles or grant additional roles later. In production environments, do not grant the Owner, Editor, or Viewer roles. Instead, grant a predefined role or custom role that meets your needs.
   7. Click Continue.

   8. Click Done to finish creating the service account.

      Do not close your browser window. You will use it in the next step.

   Create a service account key:

   1. In the Google Cloud console, click the email address for the service account that you created.
   2. Click Keys.
   3. Click Add key, and then click Create new key.
   4. Click Create. A JSON key file is downloaded to your computer.
   5. Click Close.

   Set the environment variable GOOGLE_APPLICATION_CREDENTIALS to the path of the JSON file that contains your credentials. This variable applies only to your current shell session, so if you open a new session, set the variable again.

   Example: Linux or macOS

   export GOOGLE_APPLICATION_CREDENTIALS="KEY_PATH"

   Replace KEY_PATH with the path of the JSON file that contains your credentials.

   For example:

   export GOOGLE_APPLICATION_CREDENTIALS="/home/user/Downloads/service-account-file.json"

   Example: Windows

   For PowerShell:

   $env:GOOGLE_APPLICATION_CREDENTIALS="KEY_PATH"

   Replace KEY_PATH with the path of the JSON file that contains your credentials.

   For example:

   $env:GOOGLE_APPLICATION_CREDENTIALS="C:\Users\username\Downloads\service-account-file.json"

   For command prompt:

   set GOOGLE_APPLICATION_CREDENTIALS=KEY_PATH

   Replace KEY_PATH with the path of the JSON file that contains your credentials.

   Install the Google Cloud CLI.

   Se estiver a usar um fornecedor de identidade (IdP) externo, tem primeiro de iniciar sessão na CLI gcloud com a sua identidade federada.

   Para inicializar a CLI gcloud, execute o seguinte comando:

   gcloud init

   In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Roles required to select or create a project

   • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
   • Create a project: To create a project, you need the Project Creator (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

   Verify that billing is enabled for your Google Cloud project.

   Enable the Dataproc, Compute Engine, and Cloud Storage APIs.

   Roles required to enable APIs

   To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

   Enable the APIs

   Create a service account:

   1. Ensure that you have the Create Service Accounts IAM role (roles/iam.serviceAccountCreator). Learn how to grant roles.

   2. In the Google Cloud console, go to the Create service account page.

      Go to Create service account
   3. Select your project.

   4. In the Service account name field, enter a name. The Google Cloud console fills in the Service account ID field based on this name.

      In the Service account description field, enter a description. For example, Service account for quickstart.

   5. Click Create and continue.

   6. Grant the Project > Owner role to the service account.

      To grant the role, find the Select a role list, then select Project > Owner.

      Note: The Role field affects which resources the service account can access in your project. You can revoke these roles or grant additional roles later. In production environments, do not grant the Owner, Editor, or Viewer roles. Instead, grant a predefined role or custom role that meets your needs.
   7. Click Continue.

   8. Click Done to finish creating the service account.

      Do not close your browser window. You will use it in the next step.

   Create a service account key:

   1. In the Google Cloud console, click the email address for the service account that you created.
   2. Click Keys.
   3. Click Add key, and then click Create new key.
   4. Click Create. A JSON key file is downloaded to your computer.
   5. Click Close.

   Set the environment variable GOOGLE_APPLICATION_CREDENTIALS to the path of the JSON file that contains your credentials. This variable applies only to your current shell session, so if you open a new session, set the variable again.

   Example: Linux or macOS

   export GOOGLE_APPLICATION_CREDENTIALS="KEY_PATH"

   Replace KEY_PATH with the path of the JSON file that contains your credentials.

   For example:

   export GOOGLE_APPLICATION_CREDENTIALS="/home/user/Downloads/service-account-file.json"

   Example: Windows

   For PowerShell:

   $env:GOOGLE_APPLICATION_CREDENTIALS="KEY_PATH"

   Replace KEY_PATH with the path of the JSON file that contains your credentials.

   For example:

   $env:GOOGLE_APPLICATION_CREDENTIALS="C:\Users\username\Downloads\service-account-file.json"

   For command prompt:

   set GOOGLE_APPLICATION_CREDENTIALS=KEY_PATH

   Replace KEY_PATH with the path of the JSON file that contains your credentials.

   Install the Google Cloud CLI.

   Se estiver a usar um fornecedor de identidade (IdP) externo, tem primeiro de iniciar sessão na CLI gcloud com a sua identidade federada.

   Para inicializar a CLI gcloud, execute o seguinte comando:

   gcloud init

   1. Crie um contentor do Cloud Storage. Precisa de um Cloud Storage para guardar os dados do tutorial. Se não tiver um pronto a usar, crie um novo contentor no seu projeto.

      1. In the Google Cloud console, go to the Cloud Storage Buckets page.

         Go to Buckets

      2. Click Create.
      3. On the Create a bucket page, enter your bucket information. To go to the next step, click Continue.
         1. In the Get started section, do the following:
            • Enter a globally unique name that meets the bucket naming requirements.
            • To add a bucket label, expand the Labels section (), click Add label, and specify a key and a value for your label.
         2. In the Choose where to store your data section, do the following:
            1. Select a Location type.
            2. Choose a location where your bucket's data is permanently stored from the Location type drop-down menu.
               • If you select the dual-region location type, you can also choose to enable turbo replication by using the relevant checkbox.
            3. To set up cross-bucket replication, select Add cross-bucket replication via Storage Transfer Service and follow these steps:

               Set up cross-bucket replication

               1. In the Bucket menu, select a bucket.

               2. In the Replication settings section, click Configure to configure settings for the replication job.

                  The Configure cross-bucket replication pane appears.

                  • To filter objects to replicate by object name prefix, enter a prefix that you want to include or exclude objects from, then click Add a prefix.
                  • To set a storage class for the replicated objects, select a storage class from the Storage class menu. If you skip this step, the replicated objects will use the destination bucket's storage class by default.
                  • Click Done.
         3. In the Choose how to store your data section, do the following:
            1. Select a default storage class for the bucket or Autoclass for automatic storage class management of your bucket's data.
            2. To enable hierarchical namespace, in the Optimize storage for data-intensive workloads section, select Enable hierarchical namespace on this bucket.
               Note: You cannot enable hierarchical namespace in existing buckets.
         4. In the Choose how to control access to objects section, select whether or not your bucket enforces public access prevention, and select an access control method for your bucket's objects.
            Note: You cannot change the Prevent public access setting if this setting is enforced at an organization policy.
         5. In the Choose how to protect object data section, do the following:
            • Select any of the options under Data protection that you want to set for your bucket.
              • To enable soft delete, click the Soft delete policy (For data recovery) checkbox, and specify the number of days you want to retain objects after deletion.
              • To set Object Versioning, click the Object versioning (For version control) checkbox, and specify the maximum number of versions per object and the number of days after which the noncurrent versions expire.
              • To enable the retention policy on objects and buckets, click the Retention (For compliance) checkbox, and then do the following:
                • To enable Object Retention Lock, click the Enable object retention checkbox.
                • To enable Bucket Lock, click the Set bucket retention policy checkbox, and choose a unit of time and a length of time for your retention period.
            • To choose how your object data will be encrypted, expand the Data encryption section (), and select a Data encryption method.
      4. Click Create.

   2. Defina variáveis de ambiente local. Defina variáveis de ambiente na sua máquina local. Defina o Google Cloud project-id e o nome do contentor do Cloud Storage que vai usar para este tutorial. Indique também o nome e a região de um cluster do Dataproc existente ou novo. Pode criar um cluster para usar neste tutorial no passo seguinte.

      PROJECT=project-id
      

      BUCKET_NAME=bucket-name
      

      CLUSTER=cluster-name
      

      REGION=cluster-region Example: "us-central1"
      

   3. Crie um cluster do Dataproc. Execute o comando abaixo para criar um cluster do Dataproc de nó único na zona do Compute Engine especificada.

      gcloud dataproc clusters create ${CLUSTER} \
          --project=${PROJECT} \
          --region=${REGION} \
          --single-node
      

      O comando acima instala a versão da imagem do cluster predefinida. Pode usar a flag --image-version para selecionar uma versão de imagem para o seu conjunto. Cada versão da imagem instala versões específicas dos componentes da biblioteca do Spark e do Scala. Se preparar a tarefa de contagem de palavras do Spark em Java ou Scala, vai fazer referência às versões do Spark e do Scala instaladas no cluster quando preparar o pacote de tarefas.

   4. Copie dados públicos para o seu contentor do Cloud Storage. Copie um fragmento de texto de Shakespeare de dados públicos para a pasta input do seu contentor do Cloud Storage:

      gcloud storage cp gs://pub/shakespeare/rose.txt \
          gs://${BUCKET_NAME}/input/rose.txt
      

   5. Configure um ambiente de desenvolvimento Java (Apache Maven), Scala (SBT) ou Python.

      Use o Cloud Shell. O Cloud Shell inclui ferramentas usadas neste tutorial, incluindo o Apache Maven, o Python e a CLI Google Cloud.

   Prepare a tarefa de contagem de palavras do Spark

   Selecione um separador abaixo para seguir os passos de preparação de um pacote de tarefas ou um ficheiro a enviar para o cluster. Pode preparar um dos seguintes tipos de tarefas:

   • Tarefa do Spark em Java usando o Apache Maven para criar um pacote JAR
   • Tarefa do Spark em Scala com o SBT para criar um pacote JAR
   • Tarefa do Spark em Python (PySpark)

   Java

   1. Copie o ficheiro pom.xml para a sua máquina local. O ficheiro pom.xml seguinte especifica as dependências da biblioteca Scala e Spark, às quais é atribuído um âmbito provided para indicar que o cluster do Dataproc vai fornecer estas bibliotecas no tempo de execução. O ficheiro pom.xml não especifica uma dependência do Cloud Storage porque o conetor implementa a interface HDFS padrão. Quando uma tarefa do Spark acede a ficheiros do cluster do Cloud Storage (ficheiros com URIs que começam por gs://), o sistema usa automaticamente o conetor do Cloud Storage para aceder aos ficheiros no Cloud Storage
      Verifique a versão da imagem do cluster. Substitua os marcadores de posição version no ficheiro para mostrar as versões da biblioteca do Spark e do Scala usadas pela versão da imagem do cluster. Tenha em atenção que o número do artefacto spark-core_ é o número da versão do Scala
      .major.minor

      <?xml version="1.0" encoding="UTF-8"?>
      <project xmlns="http://maven.apache.org/POM/4.0.0"
          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
      
        <groupId>dataproc.codelab</groupId>
        <artifactId>word-count</artifactId>
        <version>1.0</version>
      
        <properties>
          <maven.compiler.source>1.8</maven.compiler.source>
          <maven.compiler.target>1.8</maven.compiler.target>
        </properties>
      
        <dependencies>
          <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>Scala version, for example, 2.11.8</version>
            <scope>provided</scope>
          </dependency>
          <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_Scala major.minor.version, for example, 2.11</artifactId>
            <version>Spark version, for example, 2.3.1</version>
            <scope>provided</scope>
          </dependency>
        </dependencies>
      </project>

   2. Copie o código WordCount.java indicado abaixo para o seu computador local.
      1. Crie um conjunto de diretórios com o caminho src/main/java/dataproc/codelab:

         mkdir -p src/main/java/dataproc/codelab
         

      2. Copie WordCount.java para a sua máquina local para src/main/java/dataproc/codelab:

         cp WordCount.java src/main/java/dataproc/codelab
         

      WordCount.java é uma tarefa do Spark em Java que lê ficheiros de texto do Cloud Storage, faz uma contagem de palavras e, em seguida, escreve os resultados do ficheiro de texto no Cloud Storage.

      package dataproc.codelab;
      
      import java.util.Arrays;
      import org.apache.spark.SparkConf;
      import org.apache.spark.api.java.JavaPairRDD;
      import org.apache.spark.api.java.JavaRDD;
      import org.apache.spark.api.java.JavaSparkContext;
      import scala.Tuple2;
      
      public class WordCount {
        public static void main(String[] args) {
          if (args.length != 2) {
            throw new IllegalArgumentException("Exactly 2 arguments are required: <inputUri> <outputUri>");
          }
          String inputPath = args[0];
          String outputPath = args[1];
          JavaSparkContext sparkContext = new JavaSparkContext(new SparkConf().setAppName("Word Count"));
          JavaRDD<String> lines = sparkContext.textFile(inputPath);
          JavaRDD<String> words = lines.flatMap(
              (String line) -> Arrays.asList(line.split(" ")).iterator()
          );
          JavaPairRDD<String, Integer> wordCounts = words.mapToPair(
              (String word) -> new Tuple2<>(word, 1)
          ).reduceByKey(
              (Integer count1, Integer count2) -> count1 + count2
          );
          wordCounts.saveAsTextFile(outputPath);
        }
      }

   3. Crie o pacote.

      mvn clean package
      

      Se a compilação for bem-sucedida, é criado um target/word-count-1.0.jar.
   4. Prepare o pacote para o Cloud Storage.

      gcloud storage cp target/word-count-1.0.jar \
          gs://${BUCKET_NAME}/java/word-count-1.0.jar
      

   Scala

   1. Copie o ficheiro build.sbt para a sua máquina local. O ficheiro build.sbt seguinte especifica as dependências da biblioteca Scala e Spark, às quais é atribuído um âmbito provided para indicar que o cluster do Dataproc vai fornecer estas bibliotecas no tempo de execução. O ficheiro build.sbt não especifica uma dependência do Cloud Storage porque o conetor implementa a interface HDFS padrão. Quando uma tarefa do Spark acede a ficheiros do cluster do Cloud Storage (ficheiros com URIs que começam por gs://), o sistema usa automaticamente o conetor do Cloud Storage para aceder aos ficheiros no Cloud Storage
      Verifique a versão da imagem do cluster. Substitua os marcadores de posição version no ficheiro para mostrar as versões da biblioteca do Spark e do Scala usadas pela versão da imagem do cluster.

      scalaVersion := "Scala version, for example, 2.11.8"
      
      name := "word-count"
      organization := "dataproc.codelab"
      version := "1.0"
      
      libraryDependencies ++= Seq(
        "org.scala-lang" % "scala-library" % scalaVersion.value % "provided",
        "org.apache.spark" %% "spark-core" % "Spark version, for example, 2.3.1" % "provided"
      )

   2. Copie word-count.scala para a sua máquina local. Este é um trabalho do Spark em Java que lê ficheiros de texto do Cloud Storage, faz uma contagem de palavras e, em seguida, escreve os resultados do ficheiro de texto no Cloud Storage.

      package dataproc.codelab
      
      import org.apache.spark.SparkContext
      import org.apache.spark.SparkConf
      
      object WordCount {
        def main(args: Array[String]) {
          if (args.length != 2) {
            throw new IllegalArgumentException(
                "Exactly 2 arguments are required: <inputPath> <outputPath>")
          }
      
          val inputPath = args(0)
          val outputPath = args(1)
      
          val sc = new SparkContext(new SparkConf().setAppName("Word Count"))
          val lines = sc.textFile(inputPath)
          val words = lines.flatMap(line => line.split(" "))
          val wordCounts = words.map(word => (word, 1)).reduceByKey(_ + _)
          wordCounts.saveAsTextFile(outputPath)
        }
      }

   3. Crie o pacote.

      sbt clean package
      

      Se a compilação for bem-sucedida, é criado um target/scala-2.11/word-count_2.11-1.0.jar.
   4. Prepare o pacote para o Cloud Storage.

      gcloud storage cp target/scala-2.11/word-count_2.11-1.0.jar \
          gs://${BUCKET_NAME}/scala/word-count_2.11-1.0.jar
      

   Python

   1. Copie word-count.py para a sua máquina local. Este é um trabalho do Spark em Python que usa o PySpark para ler ficheiros de texto do Cloud Storage, faz uma contagem de palavras e, em seguida, escreve os resultados do ficheiro de texto no Cloud Storage.

      #!/usr/bin/env python
      
      import pyspark
      import sys
      
      if len(sys.argv) != 3:
        raise Exception("Exactly 2 arguments are required: <inputUri> <outputUri>")
      
      inputUri=sys.argv[1]
      outputUri=sys.argv[2]
      
      sc = pyspark.SparkContext()
      lines = sc.textFile(sys.argv[1])
      words = lines.flatMap(lambda line: line.split())
      wordCounts = words.map(lambda word: (word, 1)).reduceByKey(lambda count1, count2: count1 + count2)
      wordCounts.saveAsTextFile(sys.argv[2])

   Envie o trabalho

   Execute o seguinte comando gcloud para enviar a tarefa wordcount para o seu cluster do Dataproc.

   Java

   gcloud dataproc jobs submit spark \
       --cluster=${CLUSTER} \
       --class=dataproc.codelab.WordCount \
       --jars=gs://${BUCKET_NAME}/java/word-count-1.0.jar \
       --region=${REGION} \
       -- gs://${BUCKET_NAME}/input/ gs://${BUCKET_NAME}/output/
   

   Scala

   gcloud dataproc jobs submit spark \
       --cluster=${CLUSTER} \
       --class=dataproc.codelab.WordCount \
       --jars=gs://${BUCKET_NAME}/scala/word-count_2.11-1.0.jar \
       --region=${REGION} \
       -- gs://${BUCKET_NAME}/input/ gs://${BUCKET_NAME}/output/
   

   Python

   gcloud dataproc jobs submit pyspark word-count.py \
       --cluster=${CLUSTER} \
       --region=${REGION} \
       -- gs://${BUCKET_NAME}/input/ gs://${BUCKET_NAME}/output/
   

   Veja a saída

   Após a conclusão da tarefa, execute o seguinte comando da CLI gcloud para ver o resultado da contagem de palavras.

   gcloud storage cat gs://${BUCKET_NAME}/output/*
   

   O resultado da contagem de palavras deve ser semelhante ao seguinte:

   (a,2)
   (call,1)
   (What's,1)
   (sweet.,1)
   (we,1)
   (as,1)
   (name?,1)
   (any,1)
   (other,1)
   (rose,1)
   (smell,1)
   (name,1)
   (would,1)
   (in,1)
   (which,1)
   (That,1)
   (By,1)
   

   Limpar

   Depois de concluir o tutorial, pode limpar os recursos que criou para que deixem de usar a quota e incorrer em custos. As secções seguintes descrevem como eliminar ou desativar estes recursos.

   Elimine o projeto

   A forma mais fácil de eliminar a faturação é eliminar o projeto que criou para o tutorial.

   Para eliminar o projeto:

   Atenção: a eliminação de um projeto tem os seguintes efeitos:

   • Tudo no projeto é eliminado. Se usou um projeto existente para as tarefas neste documento, quando o elimina, também elimina qualquer outro trabalho que tenha feito no projeto.
   • Os IDs dos projetos personalizados são perdidos. Quando criou este projeto, pode ter criado um ID do projeto personalizado que quer usar no futuro. Para preservar os URLs que usam o ID do projeto, como um URL appspot.com, elimine os recursos selecionados no projeto em vez de eliminar o projeto inteiro.

   Se planeia explorar várias arquiteturas, tutoriais ou inícios rápidos, a reutilização de projetos pode ajudar a evitar exceder os limites de quota de projetos.

   1. In the Google Cloud console, go to the Manage resources page.

      Go to Manage resources

   2. In the project list, select the project that you want to delete, and then click Delete.
   3. In the dialog, type the project ID, and then click Shut down to delete the project.

   Elimine o cluster do Dataproc

   Em vez de eliminar o projeto, pode optar por eliminar apenas o cluster no projeto.

   Elimine o contentor do Cloud Storage

   In the Google Cloud console, go to the Cloud Storage Buckets page.

   Go to Buckets

   Click the checkbox for the bucket that you want to delete.

   To delete the bucket, click Delete, and then follow the instructions.

   O que se segue?

   • Veja sugestões de otimização de tarefas do Spark