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.

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 nuevos de Google Cloud califiquen para obtener una prueba gratuita.

Antes de comenzar

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

Configura tu proyecto. Si es necesario, configura un proyecto con las API de Dataproc, Compute Engine y Cloud Storage habilitadas y la CLI de Google Cloud instalada en tu máquina local.
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. En la consola de Google Cloud, ve a la página Buckets de Cloud Storage.
  Ir a la página Buckets
2. Haga clic en Crear bucket.
3. En la página Crear un bucket, ingresa la información de tu bucket. Para ir al paso siguiente, haz clic en Continuar.
  - En Nombre de tu bucket, ingresa un nombre que cumpla con los requisitos de nomenclatura de buckets.
  - En Elige dónde almacenar tus datos, haz lo siguiente:
    - Selecciona una opción de Tipo de ubicación.
    - Selecciona una opción de Ubicación.
  - Para Elegir una clase de almacenamiento predeterminada para tus datos, selecciona una clase de almacenamiento.
  - En Elige cómo controlar el acceso a los objetos, selecciona una opción de Control de acceso.
  - Para la Configuración avanzada (opcional), especifica un método de encriptación, una política de retención o etiquetas de bucket.
4. Haga clic en Crear.
Configura las variables de entorno local. Configura las variables de entorno en tu máquina local. Configura tu ID del proyecto de Google Cloud 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"
```
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 a fin de 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.
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:
```
gsutil cp gs://pub/shakespeare/rose.txt \
    gs://${BUCKET_NAME}/input/rose.txt
```
Configura un entorno de desarrollo de Java (Apache Maven), Scala (SBT) o Python.
Usa Cloud Shell Cloud Shell incluye herramientas usadas en este instructivo, incluidas Apache Maven, Python y las herramientas de línea de comandos de gcloud y gsutil de la CLI de Google Cloud.

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

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>

Copia el código WordCount.java que aparece a continuación en tu máquina local.

Crea un conjunto de directorios con la ruta src/main/java/dataproc/codelab:
```
mkdir -p src/main/java/dataproc/codelab
```
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 simple 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);
  }
}

Compila el paquete.
```
mvn clean package
```
Si la compilación se realiza de forma correcta, se crea una target/spark-with-gcs-1.0-SNAPSHOT.jar.

Almacena el paquete en etapa intermedia en Cloud Storage.

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

Scala

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"
)
```

Copia word-count.scala a tu máquina local. Este es un trabajo simple 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)
  }
}

Compila el paquete.
```
sbt clean package
```
Si la compilación se realiza de forma correcta, se crea una target/scala-2.11/word-count_2.11-1.0.jar.

Almacena el paquete en etapa intermedia en Cloud Storage.

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

Python

Copia word-count.py a tu máquina local. Este es un trabajo simple 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 gsutil de la CLI de gcloud para ver el resultado del recuento de palabras.

gsutil 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 estas 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 ID personalizados de proyectos. Cuando creaste este proyecto, es posible que hayas creado un ID del proyecto personalizado que desees utilizar en el futuro. Para conservar las URL que utilizan el ID del proyecto, como una URL appspot.com, borra los recursos seleccionados dentro del proyecto en lugar de borrar 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.

En la consola de Google Cloud, ve a la página Administrar recursos.
Ir a Administrar recursos
En la lista de proyectos, elige el proyecto que quieres borrar y haz clic en Borrar.
En el diálogo, escribe el ID del proyecto y, luego, haz clic en Cerrar para borrar el proyecto.

Cómo borrar 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

Google Cloud Console

En la consola de Google Cloud, ve a la página Navegador de Cloud Storage.
Ir al navegador
Haz clic en la casilla de verificación del bucket que deseas borrar.
Para borrar el bucket, haz clic en Borrar y sigue las instrucciones.

Línea de comandos

gsutil rb BUCKET_NAME

¿Qué sigue?

Consulta las sugerencias de ajuste de trabajo de Spark