Usa el conector de Cloud Storage con Apache Spark

---

En este instructivo, se muestra cómo ejecutar código de ejemplo que usa el conector de Cloud Storage con Apache Spark.

Objetivos

Escribe un trabajo simple de Spark de recuento de palabras en Java, Scala o Python y, luego, ejecuta el trabajo en un clúster de Dataproc.

Todas las versiones de imagen del clúster de Dataproc ya tienen instalados los componentes de Spark necesarios para este instructivo.

Costos

En este documento, usarás los siguientes componentes facturables de Google Cloud:

• Compute Engine
• Dataproc
• Cloud Storage

Para generar una estimación de costos en función del uso previsto, usa la calculadora de precios.

Es posible que los usuarios de Google Cloud nuevos cumplan con los requisitos para acceder a una prueba gratuita.

Antes de comenzar

Ejecuta los siguientes pasos para prepararte para ejecutar el código en este instructivo.

1. Configura tu proyecto. Si es necesario, configura un proyecto con las APIs de Dataproc, Compute Engine y Cloud Storage habilitadas, y Google Cloud CLI instalada en tu máquina 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.

   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.

   Enable the APIs

   Create a service account:

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

      Go to Create service account
   2. Select your project.

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

   4. Click Create and continue.

   5. 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.
   6. Click Continue.

   7. 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.

   Si usas un proveedor de identidad externo (IdP), primero debes acceder a gcloud CLI con tu identidad federada.

   Para inicializar gcloud CLI, ejecuta el siguiente comando:

   gcloud init

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

   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.

   Enable the APIs

   Create a service account:

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

      Go to Create service account
   2. Select your project.

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

   4. Click Create and continue.

   5. 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.
   6. Click Continue.

   7. 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.

   Si usas un proveedor de identidad externo (IdP), primero debes acceder a gcloud CLI con tu identidad federada.

   Para inicializar gcloud CLI, ejecuta el siguiente comando:

   gcloud init

   1. Crea un bucket de Cloud Storage. Necesitas Cloud Storage para guardar los datos del instructivo. Si no tienes uno listo para usar, crea un bucket nuevo en tu proyecto.

      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. Configura las variables de entorno local. Configura las variables de entorno en tu máquina local. Configura tu Google Cloud project-id y el nombre del bucket de Cloud Storage que usarás para este instructivo. También proporciona el nombre y la región de un clúster de Dataproc nuevo o existente. Puedes crear un clúster para usar en este instructivo en el siguiente paso.

      PROJECT=project-id
      

      BUCKET_NAME=bucket-name
      

      CLUSTER=cluster-name
      

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

   3. Crea un clúster de Dataproc. Ejecuta el comando siguiente para crear un clúster de Dataproc de nodo único en la zona de Compute Engine especificada.

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

      Con el comando anterior, se instala la versión de la imagen del clúster predeterminada. Puedes usar la marca --image-version para seleccionar una versión de imagen para tu clúster. Cada versión con imágenes instala versiones específicas de los componentes de la biblioteca de Spark y Scala. Si preparas el trabajo de recuento de palabras de Spark en Java o Scala, harás referencia a las versiones de Spark y Scala instaladas en tu clúster cuando prepares el paquete de trabajo.

   4. Copia los datos públicos en tu bucket de Cloud Storage Copia un fragmento de texto público de un fragmento de texto de Shakespeare en la carpeta input de tu bucket de Cloud Storage:

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

   5. Configura un entorno de desarrollo de Java (Apache Maven), Scala (SBT) o Python.

      Usa Cloud Shell Cloud Shell incluye las herramientas que se usan en este instructivo, como Apache Maven, Python y Google Cloud CLI.

   Prepara el trabajo de conteo de palabras de Spark

   Selecciona una pestaña, a continuación, para seguir los pasos a fin de preparar un paquete o un archivo de trabajo a fin de enviarlo a tu clúster. Puedes preparar uno de los siguientes tipos de trabajos;

   • Trabajo Spark en Java con Apache Maven para compilar un paquete de JAR
   • Spark job en Scala con SBT para compilar un paquete de JAR
   • Trabajo de Spark en Python (PySpark)

   Java

   1. Copia el archivo pom.xml en tu máquina local. En el siguiente archivo pom.xml, se especifican las dependencias de la biblioteca de Scala y Spark, que tienen un permiso provided para indicar que el clúster de Dataproc proporcionará estas bibliotecas en el entorno de ejecución. El archivo pom.xml no especifica una dependencia de Cloud Storage porque el conector implementa la interfaz de HDFS estándar. Cuando un trabajo de Spark accede a los archivos del clúster de Cloud Storage (archivos con URI que comienzan con gs://), el sistema usa de forma automática el conector de Cloud Storage para acceder a los archivos en Cloud Storage
      Verifica la versión de imagen del clúster. Reemplaza los marcadores de posición version en el archivo para mostrar las versiones de la biblioteca de Spark y Scala que usa la versión de imagen del clúster. Ten en cuenta que el número de artefacto spark-core_ es el número de versión major.minor de Scala.

      <?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. Copia el código WordCount.java que aparece a continuación en tu máquina local.
      1. Crea un conjunto de directorios con la ruta src/main/java/dataproc/codelab:

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

      2. Copia WordCount.java en tu máquina local en src/main/java/dataproc/codelab:

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

      WordCount.java es un trabajo de Spark en Java que lee archivos de texto de Cloud Storage, realiza un conteo de palabras y, luego, escribe los resultados del archivo de texto en 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. Compila el paquete.

      mvn clean package
      

      Si la compilación se realiza de forma correcta, se crea un target/word-count-1.0.jar.
   4. Almacena el paquete en etapa intermedia en Cloud Storage.

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

   Scala

   1. Copia el archivo build.sbt en tu máquina local. En el siguiente archivo build.sbt, se especifican las dependencias de la biblioteca de Scala y Spark, que tienen un permiso provided para indicar que el clúster de Dataproc proporcionará estas bibliotecas en el entorno de ejecución. El archivo build.sbt no especifica una dependencia de Cloud Storage porque el conector implementa la interfaz de HDFS estándar. Cuando un trabajo de Spark accede a los archivos del clúster de Cloud Storage (archivos con URI que comienzan con gs://), el sistema usa de forma automática el conector de Cloud Storage para acceder a los archivos en Cloud Storage
      Verifica la versión de la imagen del clúster. Reemplaza los marcadores de posición version en el archivo para mostrar las versiones de la biblioteca de Spark y Scala que usa la versión de imagen de tu clúster.

      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. Copia word-count.scala a tu máquina local. Este es un trabajo de Spark en Java que lee archivos de texto de Cloud Storage, realiza un conteo de palabras y, luego, escribe los resultados del archivo de texto en 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. Compila el paquete.

      sbt clean package
      

      Si la compilación se realiza de forma correcta, se crea un target/scala-2.11/word-count_2.11-1.0.jar.
   4. Almacena el paquete en etapa intermedia en 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. Copia word-count.py a tu máquina local. Este es un trabajo de Spark en Python con PySpark que lee archivos de texto de Cloud Storage, realiza un conteo de palabras y, luego, escribe los resultados del archivo de texto en 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])

   Envía el trabajo

   Ejecuta el siguiente comando de gcloud para enviar el trabajo de conteo de palabras a tu clúster de 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/
   

   Revise el resultado.

   Una vez que finalice el trabajo, ejecuta el siguiente comando de gcloud CLI para ver el resultado del recuento de palabras.

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

   El resultado del conteo de palabras debe ser similar al siguiente:

   (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)
   

   Limpia

   Una vez que completes el instructivo, puedes limpiar los recursos que creaste para que dejen de usar la cuota y generar cargos. En las siguientes secciones, se describe cómo borrar o desactivar estos recursos.

   Borra el proyecto

   La manera más fácil de eliminar la facturación es borrar el proyecto que creaste para el instructivo.

   Para borrar el proyecto, sigue estos pasos:

   Precaución: Borrar un proyecto tiene las siguientes consecuencias:

   • Se borra todo en el proyecto. Si usaste un proyecto existente para las tareas de este documento, cuando lo borres, también se borrará cualquier otro trabajo que hayas realizado en el proyecto.
   • Se pierden los IDs personalizados de los proyectos. Cuando creaste este proyecto, es posible que hayas creado un ID del proyecto personalizado que desees utilizar en el futuro. Para conservar las URLs que usan el ID del proyecto, como una URL appspot.com, borra los recursos seleccionados dentro del proyecto, en lugar de todo el proyecto.

   Si planeas explorar varias infraestructuras, instructivos y guías de inicio rápido la reutilización de proyectos puede ayudarte a evitar exceder los límites de las cuotas del proyecto.

   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.

   Borra el clúster de Dataproc.

   En lugar de borrar tu proyecto, es posible que solo quieras borrar tu clúster dentro del proyecto.

   Borra el bucket de 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.

   ¿Qué sigue?

   • Consulta las sugerencias de ajuste de trabajo de Spark