Escribe y ejecuta trabajos de Spark Scala en Dataproc

En este instructivo, se ilustran diferentes formas de crear y enviar un trabajo de Spark Scala a un clúster de Dataproc, lo que incluye cómo:

  • escribir y compilar una aplicación "Hello World" de Spark Scala en una máquina local desde la línea de comandos con el REPL de Scala (Read-Evaluate-Print-Loop o intérprete interactivo) o la herramienta de compilación SBT
  • empaquetar las clases de Scala compiladas en un archivo jar con un manifiesto
  • enviar el jar de Scala a un trabajo de Spark que se ejecuta en tu clúster de Dataproc
  • examinar el resultado del trabajo de Scala desde la consola de Google Cloud

En este instructivo, también se muestra cómo:

  • escribir y ejecutar un trabajo de MapReduce "WordCount" de Spark Scala directamente en un clúster de Dataproc con el REPL spark-shell

  • ejecutar ejemplos preinstalados de Apache Spark y Hadoop en un clúster

Configura un proyecto de Google Cloud Platform

Si aún no lo hiciste, sigue estos pasos:

  1. Configura un proyecto
  2. Crear un bucket de Cloud Storage
  3. Cree un clúster de Dataproc.

Escribe y compila el código de Scala de forma local

Como un ejercicio simple para este instructivo, escribe “Hello World” app de Scala con Scala REPL o la interfaz de línea de comandos SBT de forma local en tu máquina de desarrollo.

Usa Scala

  1. Descarga los objetos binarios de Scala desde la página de instalación de Scala

  2. Descomprime el archivo, establece la variable de entorno SCALA_HOME y agrégala a tu ruta de acceso, como se muestra en la Instalación de Scala instrucciones. Por ejemplo: 

    export SCALA_HOME=/usr/local/share/scala
export PATH=$PATH:$SCALA_HOME/

  3. Inicia el REPL de Scala 

    $ scala
Welcome to Scala version ...
Type in expressions to have them evaluated.
Type :help for more information.
scala>

  4. Copia y pega el código HelloWorld en el REPL de Scala. 

    object HelloWorld {
  def main(args: Array[String]): Unit = {
    println("Hello, world!")
  }
}

  5. Guarda HelloWorld.scala y sale del REPL 

    scala> :save HelloWorld.scala
scala> :q

  6. Compila con scalac 

    $ scalac HelloWorld.scala

  7. Enumera los archivos .class compilados 

    $ ls HelloWorld*.class
HelloWorld$.class   HelloWorld.class

Usar SBT

  1. Descarga SBT

  2. Crea un proyecto "HelloWorld", como se muestra a continuación 

    $ mkdir hello
$ cd hello
$ echo \
'object HelloWorld {def main(args: Array[String]) = println("Hello, world!")}' > \
HelloWorld.scala

  3. Crea un archivo de configuración sbt.build para establecer artifactName (el nombre de la archivo jar que generarás a continuación) a “HelloWorld.jar” (consulta Modifica los artefactos predeterminados) 

    echo \
'artifactName := { (sv: ScalaVersion, module: ModuleID, artifact: Artifact) =>
"HelloWorld.jar" }' > \
build.sbt

  4. Inicia SBT y ejecuta el código 

    $ sbt
[info] Set current project to hello ...
> run
... Compiling 1 Scala source to .../hello/target/scala-.../classes...
... Running HelloWorld
Hello, world!
[success] Total time: 3 s ...

  5. Empaqueta código en un archivo jar con un manifiesto que especifica el punto de entrada de la clase principal (HelloWorld) y, luego, sal.

    > package
... Packaging .../hello/target/scala-.../HelloWorld.jar ...
... Done packaging.
[success] Total time: ...
> exit

Crea un jar

Crea un archivo jar con SBT o con el comando jar.

Crea un jar con SBT

El comando package de SBT crea un archivo jar (consulta Cómo usar SBT).

Crea un archivo jar manualmente

  1. Cambia el directorio (cd) al directorio que contiene tus archivos HelloWorld*.class compilados y, luego, ejecuta el siguiente comando para empaquetar los archivos de clase en un jar con un manifiesto que especifique el punto de entrada de la clase principal (HelloWorld).
    $ jar cvfe HelloWorld.jar HelloWorld HelloWorld*.class
added manifest
adding: HelloWorld$.class(in = 637) (out= 403)(deflated 36%)
adding: HelloWorld.class(in = 586) (out= 482)(deflated 17%)

Copia el jar a Cloud Storage

  1. Usa Google Cloud CLI para copiar el jar a un bucket de Cloud Storage en tu proyecto
$ gcloud storage cp HelloWorld.jar gs://<bucket-name>/
Copying file://HelloWorld.jar [Content-Type=application/java-archive]...
Uploading   gs://bucket-name/HelloWorld.jar:         1.46 KiB/1.46 KiB

Envía el jar a un trabajo de Spark de Dataproc

  1. Usa la consola de Google Cloud para enviar el archivo jar a tu trabajo de Spark de Dataproc. Completa los campos en la página Enviar un trabajo de la siguiente manera:

    • Clúster: Selecciona el nombre de tu clúster de la lista de clústeres
    • Tipo de trabajo: Spark

    • Clase o archivo JAR principal: Especifica la ruta de acceso del URI de Cloud Storage a tu archivo JAR de HelloWorld (gs://your-bucket-name/HelloWorld.jar).

      Si tu archivo JAR no incluye un manifiesto que especifique el punto de entrada a tu código ("Main-Class: HelloWorld"), el campo "Main class or jar" debe indicar el nombre de tu clase principal ("HelloWorld") y debes completar el campo "Jar files" con la ruta de acceso del URI a tu archivo JAR (gs://your-bucket-name/HelloWorld.jar).

  2. Haz clic en Submit (Enviar) para iniciar el trabajo. Cuando se inicia el trabajo, este se agrega a la lista de objetos Job lista.

  3. Haz clic en el ID de trabajo para abrir la página Jobs, en la que podrás ver el resultado del controlador del trabajo.

Escribe y ejecuta el código de Spark Scala con el REPL spark-shell del clúster

Te recomendamos que desarrolles aplicaciones de Scala directamente en tu clúster de Dataproc. Hadoop y Spark son preinstalados en los clústeres de Dataproc y se configuran con el Cloud Storage que permite que tu código lea y escriba datos directamente desde y hacia Cloud Storage.

En este ejemplo, se muestra cómo establecer una conexión SSH al nodo principal del clúster de Dataproc de tu proyecto y, luego, usar el REPL spark-shell para crear y ejecutar una aplicación MapReduce WordCount de Scala.

  1. Establece una conexión SSH al nodo principal del clúster de Dataproc de la siguiente manera:

    1. Ve a la página Clústeres de Dataproc de tu proyecto en la consola de Google Cloud y, luego, haz clic en el nombre de tu clúster.

    2. En la página de detalles del clúster, selecciona la pestaña Instancias de VM y, luego, haz clic en Selección de SSH que aparece a la derecha de la fila del nombre del clúster.

      Se abre una ventana del navegador en tu directorio principal del nodo principal.

  2. Inicia el spark-shell 

    $ spark-shell
...
Using Scala version ...
Type in expressions to have them evaluated.
Type :help for more information.
...
Spark context available as sc.
...
SQL context available as sqlContext.
scala>

  3. Crea un RDD (conjunto de datos resilientes y distribuidos) a partir de un fragmento de texto de Shakespeare ubicado en un almacenamiento público de Cloud Storage 

    scala> val text_file = sc.textFile("gs://pub/shakespeare/rose.txt")

  4. Ejecuta un MapReduce de WordCount en el texto y, luego, muestra el resultado wordcounts 

    scala> val wordCounts = text_file.flatMap(line => line.split(" ")).map(word =>
(word, 1)).reduceByKey((a, b) => a + b)
scala> wordCounts.collect
... Array((call,1), (What's,1), (sweet.,1), (we,1), (as,1), (name?,1), (any,1), (other,1),
(rose,1), (smell,1), (name,1), (a,2), (would,1), (in,1), (which,1), (That,1), (By,1))

  5. Guarda los recuentos en <bucket-name>/wordcounts-out en Cloud Storage y, luego, sal de scala-shell 

    scala> wordCounts.saveAsTextFile("gs://<bucket-name>/wordcounts-out/")
scala> exit

  6. Usa gcloud CLI para crear una lista de los archivos de salida y mostrar su contenido 

    $ gcloud storage ls gs://bucket-name/wordcounts-out/
gs://spark-scala-demo-bucket/wordcounts-out/
gs://spark-scala-demo-bucket/wordcounts-out/_SUCCESS
gs://spark-scala-demo-bucket/wordcounts-out/part-00000
gs://spark-scala-demo-bucket/wordcounts-out/part-00001

  7. Comprueba los contenidos de gs://<bucket-name>/wordcounts-out/part-00000 

    $ gcloud storage cat gs://bucket-name/wordcounts-out/part-00000
(call,1)
(What's,1)
(sweet.,1)
(we,1)
(as,1)
(name?,1)
(any,1)
(other,1)

Ejecuta el código de ejemplo preinstalado

El nodo principal de Dataproc contiene archivos jar ejecutables con Apache Hadoop estándar y Spark. ejemplos.

Tipo de jar Master node /usr/lib/ location Fuente de GitHub Documentos de Apache
Hadoop hadoop-mapreduce/hadoop-mapreduce-examples.jar vínculo fuente Instructivo de MapReduce
Spark spark/lib/spark-examples.jar vínculo fuente Ejemplos de Spark

Envía ejemplos a tu clúster desde la línea de comandos

Puedes enviar ejemplos desde tu máquina de desarrollo local con la herramienta de línea de comandos de gcloud de Google Cloud CLI (consulta la documentación sobre cómo usar la consola de Google Cloud para enviar trabajos desde la consola de Google Cloud).

Ejemplo de WordCount de Hadoop

gcloud dataproc jobs submit hadoop --cluster=cluster-name \
    --region=region \
    --jars=file:///usr/lib/hadoop-mapreduce/hadoop-mapreduce-examples.jar \
    --class=org.apache.hadoop.examples.WordCount \
    -- URI of input file URI of output file

Ejemplo de WordCount de Spark

gcloud dataproc jobs submit spark --cluster=cluster-name \
    --region=region \
    --jars=file:///usr/lib/spark/examples/jars/spark-examples.jar \
    --class=org.apache.spark.examples.JavaWordCount \
    -- URI of input file

Cierra el clúster

Para evitar cargos continuos, cierra el clúster y borra los recursos de Cloud Storage (el bucket y los archivos de Cloud Storage) que se usaron para este instructivo.

Para cerrar un clúster: 

gcloud dataproc clusters delete cluster-name \
    --region=region

Para borrar el archivo jar de Cloud Storage: 

gcloud storage rm gs://bucket-name/HelloWorld.jar

Puedes borrar un bucket y todas sus carpetas y archivos con el siguiente comando: 

gcloud storage rm gs://bucket-name/ --recursive

¿Qué sigue?