Utilizar el conector de Cloud Storage con Apache Spark

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

Objetivos

Escribe una tarea sencilla de recuento de palabras de Spark en Java, Scala o Python y, a continuación, ejecuta la tarea en un clúster de Dataproc.

Costes

En este documento, se utilizan los siguientes componentes facturables de Google Cloud:

  • Compute Engine
  • Dataproc
  • Cloud Storage

Para generar una estimación de costes basada en el uso previsto, utiliza la calculadora de precios.

Los usuarios nuevos Google Cloud pueden disfrutar de una prueba gratuita.

Antes de empezar

Sigue los pasos que se indican a continuación para preparar la ejecución del código de este tutorial.

  1. Configura tu proyecto. Si es necesario, configura un proyecto con las APIs de Dataproc, Compute Engine y Cloud Storage habilitadas, y la CLI de Google Cloud instalada en tu máquina local.

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

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

      Go to project selector

    3. Verify that billing is enabled for your Google Cloud project.

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

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

      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.

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

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

    8. Install the Google Cloud CLI.

    9. Si utilizas un proveedor de identidades (IdP) externo, primero debes iniciar sesión en la CLI de gcloud con tu identidad federada.

    10. Para inicializar gcloud CLI, ejecuta el siguiente comando: 

      gcloud init

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

      Go to project selector

    12. Verify that billing is enabled for your Google Cloud project.

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

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

      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.

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

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

    17. Install the Google Cloud CLI.

    18. Si utilizas un proveedor de identidades (IdP) externo, primero debes iniciar sesión en la CLI de gcloud con tu identidad federada.

    19. Para inicializar gcloud CLI, ejecuta el siguiente comando: 

      gcloud init

    20. Crea un segmento de Cloud Storage. Necesitas un segmento de Cloud Storage para almacenar los datos del tutorial. Si no tienes ninguno listo para usar, crea un nuevo contenedor 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.
          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.
        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.
        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.

    21. Define variables de entorno locales. Define variables de entorno en tu máquina local. Define el Google Cloud project-id y el nombre del segmento de Cloud Storage que usarás en este tutorial. También debes proporcionar el nombre y la región de un clúster de Dataproc nuevo o que ya tengas. En el siguiente paso, puedes crear un clúster para usarlo en este tutorial. 

      PROJECT=project-id

      BUCKET_NAME=bucket-name

      CLUSTER=cluster-name

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

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

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

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

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

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

      25. Preparar la tarea de recuento de palabras de Spark

      Selecciona una pestaña de abajo para seguir los pasos para preparar un paquete de trabajo o un archivo que quieras enviar a tu clúster. Puedes preparar uno de los siguientes tipos de trabajo:

      Java

      1. Copia el archivo pom.xml en tu equipo local. El siguiente archivo pom.xml especifica las dependencias de la biblioteca de Scala y Spark, a las que se les asigna un ámbito provided para indicar que el clúster de Dataproc proporcionará estas bibliotecas en el tiempo de ejecución. El archivo pom.xml no especifica una dependencia de Cloud Storage porque el conector implementa la interfaz HDFS estándar. Cuando una tarea de Spark accede a archivos de clúster de Cloud Storage (archivos con URIs que empiezan por gs://), el sistema utiliza automáticamente el conector de Cloud Storage para acceder a los archivos de Cloud Storage. 
        <?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 se indica más abajo en tu equipo 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 una tarea de Spark en Java que lee archivos de texto de Cloud Storage, realiza un recuento de palabras y, a continuación, 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. Crea el paquete. 
        mvn clean package
        Si la compilación se realiza correctamente, se crea un target/word-count-1.0.jar.
      4. Transfiere el paquete a 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 equipo local. El siguiente archivo build.sbt especifica las dependencias de la biblioteca de Scala y Spark, a las que se les asigna un ámbito provided para indicar que el clúster de Dataproc proporcionará estas bibliotecas en el tiempo de ejecución. El archivo build.sbt no especifica una dependencia de Cloud Storage porque el conector implementa la interfaz HDFS estándar. Cuando un trabajo de Spark accede a archivos de clúster de Cloud Storage (archivos con URIs que empiezan por gs://), el sistema utiliza automáticamente el conector de Cloud Storage para acceder a los archivos de Cloud Storage. 
        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 en tu equipo local. Se trata de un trabajo de Spark en Java que lee archivos de texto de Cloud Storage, realiza un recuento de palabras y, a continuación, 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. Crea el paquete. 
        sbt clean package
        Si la compilación se realiza correctamente, se crea un target/scala-2.11/word-count_2.11-1.0.jar.
      4. Transfiere el paquete a 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 en tu equipo local. Se trata de un trabajo de Spark en Python que usa PySpark para leer archivos de texto de Cloud Storage, realizar un recuento de palabras y, a continuación, escribir 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])

      Enviar la tarea

      Ejecuta el siguiente comando gcloud para enviar la tarea de recuento 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/

      Ver el resultado

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

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

      El resultado del recuento de palabras debería 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)

      Limpieza

      Cuando hayas terminado el tutorial, puedes eliminar los recursos que has creado para que dejen de usar cuota y generar cargos. En las siguientes secciones se explica cómo eliminar o desactivar dichos recursos.

      Eliminar el proyecto

      La forma más fácil de evitar que te cobren es eliminar el proyecto que has creado para el tutorial.

      Para ello, sigue las instrucciones que aparecen a continuación:

      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.

      Eliminar el clúster de Dataproc

      En lugar de eliminar el proyecto, puedes eliminar solo el clúster del proyecto.

      Eliminar el segmento de Cloud Storage

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

      Go to Buckets

    26. Click the checkbox for the bucket that you want to delete.
    27. To delete the bucket, click Delete, and then follow the instructions.

      28. Siguientes pasos