Usa el conector de BigQuery con Spark

El spark-bigquery-connector se usa con Apache Spark para leer y escribir datos desde y hacia BigQuery. En este instructivo, se proporciona un código de ejemplo que usa el conector spark-bigquery dentro de una aplicación de Spark. Si deseas obtener instrucciones para crear un clúster, consulta el Guías de inicio rápido de Dataproc.

Haz que el conector esté disponible para tu aplicación

Puedes hacer que el conector spark-bigquery esté disponible para tu aplicación de una de las siguientes maneras:

  1. Instalar el conector spark-bigquery en el directorio jars de Spark de cada con el comando Acción de inicialización de los conectores de Dataproc cuando creas tu clúster.

  2. Proporciona el URI del conector cuando envíes tu trabajo:

    1. Consola de Google Cloud: Usa el elemento Jars files del trabajo de Spark en la página Enviar un trabajo de Dataproc.
    2. gcloud CLI: Usa la marca gcloud dataproc jobs submit spark --jars.
    3. API de Dataproc: usa el Campo SparkJob.jarFileUris.

  3. Incluye el jar en tu aplicación de Scala o Java Spark como dependencia (consulta Compilar con el conector).

Cómo especificar el URI del archivo jar del conector

Las versiones del conector de BigQuery de Spark se enumeran en el repositorio de GoogleCloudDataproc/spark-bigquery-connector de GitHub.

Especifica el jar del conector reemplazando la versión de Scala y del conector. información en la siguiente cadena de URI:

gs://spark-lib/bigquery/spark-bigquery-with-dependencies_SCALA_VERSION-CONNECTOR_VERSION.jar

  • Usa 2.12 de Scala con las versiones de imágenes 1.5+ de Dataproc. 

    gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-CONNECTOR_VERSION.jar

    Ejemplo de gcloud CLI: 

    gcloud dataproc jobs submit spark \
    --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.23.2.jar \
    -- job-args

  • Usa 2.11 de Scala con las versiones de imagen 1.4 y anteriores de Dataproc: 

    gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.11-CONNECTOR_VERSION.jar

    Ejemplo de gcloud CLI: 

    gcloud dataproc jobs submit spark \
    --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.11-0.23.2.jar \
    -- job-args

Calcula los costos

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

  • Dataproc
  • BigQuery
  • 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.

Cómo leer y escribir datos desde BigQuery

Este ejemplo lee los datos de BigQuery en un DataFrame de Spark para realizar un recuento de palabras mediante la API de fuente de datos estándar.

El conector escribe los datos en BigQuery almacenar en búfer todos los datos en una tabla temporal de Cloud Storage. Luego, copia todos los datos a BigQuery en una sola operación. El conector intenta borrar los archivos temporales una vez que la operación de carga de BigQuery se realiza correctamente, y lo vuelve a hacer cuando la aplicación Spark finaliza. Si el trabajo falla, quita las tareas temporales restantes Archivos de Cloud Storage. Por lo general, los eventos temporales de BigQuery se encuentran en gs://[bucket]/.spark-bigquery-[jobid]-[UUID].

Cómo configurar la facturación

De forma predeterminada, el proyecto asociado con las credenciales o la cuenta de servicio está factura por el uso de la API. Para facturar un proyecto diferente, establece la siguiente configuración: spark.conf.set("parentProject", "<BILLED-GCP-PROJECT>").

También se puede agregar a una operación de lectura o escritura, de la siguiente manera: .option("parentProject", "<BILLED-GCP-PROJECT>").

Ejecuta el código

Antes de ejecutar este ejemplo, crea un conjunto de datos llamado “wordcount_dataset” o cambia el conjunto de datos de salida en el código a un conjunto de datos de BigQuery existente en tu proyecto de Google Cloud.

Usa el comando de bq para crear el wordcount_dataset:

bq mk wordcount_dataset

Usa el comando de Google Cloud CLI para crear un bucket de Cloud Storage, que se usará para exportar a BigQuery:

gcloud storage buckets create gs://[bucket]

Scala

  1. Examina el código y reemplaza el marcador de posición [bucket] por en el bucket de Cloud Storage que creaste anteriormente. 
    /*
 * Remove comment if you are not running in spark-shell.
 *
import org.apache.spark.sql.SparkSession
val spark = SparkSession.builder()
  .appName("spark-bigquery-demo")
  .getOrCreate()
*/

// Use the Cloud Storage bucket for temporary BigQuery export data used
// by the connector.
val bucket = "[bucket]"
spark.conf.set("temporaryGcsBucket", bucket)

// Load data in from BigQuery. See
// https://github.com/GoogleCloudDataproc/spark-bigquery-connector/tree/0.17.3#properties
// for option information.
val wordsDF =
  (spark.read.format("bigquery")
  .option("table","bigquery-public-data:samples.shakespeare")
  .load()
  .cache())

wordsDF.createOrReplaceTempView("words")

// Perform word count.
val wordCountDF = spark.sql(
  "SELECT word, SUM(word_count) AS word_count FROM words GROUP BY word")
wordCountDF.show()
wordCountDF.printSchema()

// Saving the data to BigQuery.
(wordCountDF.write.format("bigquery")
  .option("table","wordcount_dataset.wordcount_output")
  .save())
  2. Ejecuta el código en tu clúster
    1. Usa SSH para conectarte al nodo instancia principal del clúster de Dataproc
      1. Ve a la sección Clústeres de Dataproc de la consola de Google Cloud y, luego, haz clic en el nombre de tu clúster
        Página de clústeres de Dataproc en la consola de Cloud.
      2. En la página >Detalles del clúster, selecciona la pestaña Instancias de VM. Luego, haz clic en SSH a la derecha del nombre del nodo de la instancia principal del clúster
        Página de detalles del clúster de Dataproc en la consola de Cloud.
        .
        Se abrirá una ventana del navegador en tu directorio principal del nodo principal. 
            Connected, host fingerprint: ssh-rsa 2048 ...
    ...
    user@clusterName-m:~$
    2. Crea wordcount.scala con el vi preinstalado. vim, o nano, editor de texto y, luego, pega en Scala código de la Lista de código de escalar 
      nano wordcount.scala
    3. Inicia el REPL spark-shell. 
      $ spark-shell --jars=gs://spark-lib/bigquery/spark-bigquery-latest.jar
...
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>
    4. Ejecuta wordcount.scala con el comando :load wordcount.scala para crear la tabla de BigQuery wordcount_output. La lista de salida muestra 20 líneas del resultado del recuento de palabras. 
      :load wordcount.scala
...
+---------+----------+
|     word|word_count|
+---------+----------+
|     XVII|         2|
|    spoil|        28|
|    Drink|         7|
|forgetful|         5|
|   Cannot|        46|
|    cures|        10|
|   harder|        13|
|  tresses|         3|
|      few|        62|
|  steel'd|         5|
| tripping|         7|
|   travel|        35|
|   ransom|        55|
|     hope|       366|
|       By|       816|
|     some|      1169|
|    those|       508|
|    still|       567|
|      art|       893|
|    feign|        10|
+---------+----------+
only showing top 20 rows

root
 |-- word: string (nullable = false)
 |-- word_count: long (nullable = true)

      Para obtener una vista previa de la tabla de resultados, abre la BigQuery selecciona la tabla wordcount_output y, luego, haz clic Vista previa.
      Vista previa de la tabla en la página del Explorador de BigQuery en la consola de Cloud.

PySpark

  1. Examina el código y reemplaza el marcador de posición [bucket] por en el bucket de Cloud Storage que creaste anteriormente. 
    #!/usr/bin/env python

"""BigQuery I/O PySpark example."""

from pyspark.sql import SparkSession

spark = SparkSession \
  .builder \
  .master('yarn') \
  .appName('spark-bigquery-demo') \
  .getOrCreate()

# Use the Cloud Storage bucket for temporary BigQuery export data used
# by the connector.
bucket = "[bucket]"
spark.conf.set('temporaryGcsBucket', bucket)

# Load data from BigQuery.
words = spark.read.format('bigquery') \
  .option('table', 'bigquery-public-data:samples.shakespeare') \
  .load()
words.createOrReplaceTempView('words')

# Perform word count.
word_count = spark.sql(
    'SELECT word, SUM(word_count) AS word_count FROM words GROUP BY word')
word_count.show()
word_count.printSchema()

# Save the data to BigQuery
word_count.write.format('bigquery') \
  .option('table', 'wordcount_dataset.wordcount_output') \
  .save()
  2. Ejecuta el código en tu clúster
    1. Usa SSH para conectarte al nodo instancia principal del clúster de Dataproc
      1. Ve a la sección Clústeres de Dataproc de la consola de Google Cloud y, luego, haz clic en el nombre de tu clúster
        Página Clústeres en la consola de Cloud.
      2. En la página Detalles del clúster, selecciona la pestaña Instancias de VM. Luego, haz clic en SSH a la derecha del nombre del nodo de la instancia principal del clúster
        Selecciona SSH en la fila del nombre del clúster en la página Detalles del clúster en la consola de Cloud.
        .
        Se abrirá una ventana del navegador en tu directorio principal del nodo principal. 
            Connected, host fingerprint: ssh-rsa 2048 ...
    ...
    user@clusterName-m:~$
    2. Crea wordcount.py con el vi preinstalado. Editor de texto vim o nano; luego, pégalo en PySpark código de la Lista de código de PySpark 
      nano wordcount.py
    3. Ejecuta el conteo de palabras con spark-submit para crear la tabla de BigQuery wordcount_output. La lista de salida muestra 20 líneas del resultado del recuento de palabras. 
      spark-submit --jars gs://spark-lib/bigquery/spark-bigquery-latest.jar wordcount.py
...
+---------+----------+
|     word|word_count|
+---------+----------+
|     XVII|         2|
|    spoil|        28|
|    Drink|         7|
|forgetful|         5|
|   Cannot|        46|
|    cures|        10|
|   harder|        13|
|  tresses|         3|
|      few|        62|
|  steel'd|         5|
| tripping|         7|
|   travel|        35|
|   ransom|        55|
|     hope|       366|
|       By|       816|
|     some|      1169|
|    those|       508|
|    still|       567|
|      art|       893|
|    feign|        10|
+---------+----------+
only showing top 20 rows

root
 |-- word: string (nullable = false)
 |-- word_count: long (nullable = true)

      Para obtener una vista previa de la tabla de resultados, abre la BigQuery selecciona la tabla wordcount_output y, luego, haz clic Vista previa.
      Vista previa de la tabla en la página del Explorador de BigQuery en la consola de Cloud.

Más información