"Managed Service for Apache Spark" is the new name for the product formerly known as "Dataproc on Compute Engine" (cluster deployment) and "Google Cloud Serverless for Apache Spark" (serverless deployment).

Google utiliza tecnología de IA para traducir contenido a tu idioma preferido. Las traducciones realizadas con IA pueden contener errores.

Usa el conector de BigQuery para Spark

El conector spark-bigquery se usa con Apache Spark para leer y escribir datos desde y hacia BigQuery. El conector aprovecha la API de BigQuery Storage cuando lee datos de BigQuery.

En este instructivo, se proporciona información sobre la disponibilidad del conector preinstalado y se muestra cómo hacer que una versión específica del conector esté disponible para los trabajos de Spark. En el código de ejemplo, se muestra cómo usar el conector de BigQuery para Spark en una aplicación de Spark.

Usa el conector preinstalado

El conector de Spark a BigQuery está preinstalado y disponible para los trabajos de Spark que se ejecutan en clústeres de Managed Service para Apache Spark creados con las versiones de imagen 2.1 y posteriores. La versión del conector preinstalada se indica en las páginas de versiones de imágenes.

Haz que una versión específica del conector esté disponible para los trabajos de Spark

Si deseas usar una versión del conector que sea diferente de una versión preinstalada en un clúster de versión de imagen 2.1 o posterior, o si deseas instalar el conector en un clúster de versión de imagen anterior a 2.1, sigue las instrucciones de esta sección.

Importante: La versión de spark-bigquery-connector debe ser compatible con la versión de la imagen del clúster de Managed Service para Apache Spark. Consulta la matriz de compatibilidad de imágenes del conector para Managed Service para Apache Spark.

Clústeres con versión de imagen `2.1` y posteriores

Cuando creas un clúster de Managed Service para Apache Spark con una versión de imagen 2.1 o posterior, especifica la versión del conector como metadatos del clúster.

Ejemplo de gcloud CLI:

gcloud dataproc clusters create CLUSTER_NAME \
    --region=REGION \
    --image-version=2.2 \
    --metadata=SPARK_BQ_CONNECTOR_VERSION or SPARK_BQ_CONNECTOR_URL\
    other flags

Notas:

SPARK_BQ_CONNECTOR_VERSION: Especifica una versión del conector. Las versiones del conector spark-bigquery se enumeran en la página spark-bigquery-connector/releases de GitHub.

Ejemplo:
```
--metadata=SPARK_BQ_CONNECTOR_VERSION=0.42.1
```
SPARK_BQ_CONNECTOR_URL: Especifica una URL que apunte al archivo JAR en Cloud Storage. Puedes especificar la URL de un conector que aparece en la columna vínculo en Descarga y usa el conector en GitHub o la ruta de acceso a una ubicación de Cloud Storage en la que colocaste un archivo JAR de conector personalizado.

Ejemplos:
```
--metadata=SPARK_BQ_CONNECTOR_URL=gs://spark-lib/bigquery/spark-3.5-bigquery-0.42.1.jar
--metadata=SPARK_BQ_CONNECTOR_URL=gs://PATH_TO_CUSTOM_JAR
```

Clústeres con versiones de imagen `2.0` y anteriores

Puedes hacer que el conector de Spark BigQuery esté disponible para tu aplicación de una de las siguientes maneras:

Instala el conector spark-bigquery en el directorio de archivos JAR de Spark de cada nodo con la acción de inicialización de conectores de Managed Service para Apache Spark cuando crees el clúster.
Proporciona la URL del archivo JAR del conector cuando envíes tu trabajo al clúster con la consola de Google Cloud , gcloud CLI o la API de Managed Service para Apache Spark.

Console

Usa el elemento Archivos JAR del trabajo de Spark en la página Enviar un trabajo de Managed Service para Apache Spark.

gcloud

Usa la marca gcloud dataproc jobs submit spark --jars.

API

Usa el campo SparkJob.jarFileUris.

Cómo especificar el archivo JAR del conector cuando se ejecutan trabajos de Spark en clústeres con versiones de imagen anteriores a la 2.0

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 sustituyendo la información de la versión de Scala y del conector en la siguiente cadena de URI:
```
gs://spark-lib/bigquery/spark-bigquery-with-dependencies_SCALA_VERSION-CONNECTOR_VERSION.jar
```
- Usa Scala 2.12 con las versiones de imagen de Managed Service para Apache Spark 1.5+
```
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 Scala 2.11 con las versiones de imagen de Managed Service para Apache Spark 1.4 y anteriores:
```
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
```
  Para un uso que no sea de producción, también puedes apuntar a los archivos JAR de latest, de la siguiente manera:
  1. Versión de la imagen de Managed Service para Apache Spark 1.5+: --jars=gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar
  2. Versiones de imagen 1.4 y anteriores de Managed Service para Apache Spark: --jars=gs://spark-lib/bigquery/spark-bigquery-latest.jar
Incluye el archivo JAR del conector en tu aplicación de Scala o Java Spark como una dependencia (consulta Compila con el conector).

Calcula los costos

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

Managed Service for Apache Spark
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 cumplan con los requisitos para acceder a una prueba gratuita.

Lee y escribe datos desde y hacia 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 lee datos de BigQuery con la API de BigQuery Storage Read, que lee directamente de los archivos de datos de la tabla de forma distribuida.

El conector escribe los datos en BigQuery en uno de dos modos: DIRECT o INDIRECT.

DIRECT: Usa la API de BigQuery Storage Write. Cada ejecutor escribe sus datos y el controlador confirma todos los datos pendientes en la tabla.
INDIRECT: Almacena en búfer los datos en una tabla temporal de Cloud Storage y, luego, activa un trabajo de carga de BigQuery para copiar los datos en BigQuery en una sola operación.

Cuando se usa el modo INDIRECT, el conector intenta borrar los archivos temporales una vez que la operación de carga de BigQuery se realiza correctamente y, de nuevo, cuando finaliza la aplicación de Spark. Si el trabajo falla, quita de forma manual los archivos temporales de Cloud Storage restantes. Por lo general, los archivos temporales de BigQuery se encuentran en gs://[bucket]/.spark-bigquery-[jobid]-[UUID].

Configura la facturación

De forma predeterminada, se factura el uso de la API al proyecto asociado con las credenciales o la cuenta de servicio. 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 por un conjunto de datos de BigQuery existente en tu proyecto deGoogle Cloud .

Usa el comando de bq para crear el wordcount_dataset:

bq mk wordcount_dataset

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

gcloud storage buckets create gs://[bucket]

Scala

Examina el código y reemplaza el marcador de posición [bucket] por el bucket de Cloud Storage que creaste antes.

/*
 * 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)
// Enable reading from BigQuery views and materializing query results.
spark.conf.set("viewsEnabled", "true")
spark.conf.set("materializationDataset", "wordcount_dataset")

// Load data in from BigQuery. See
// https://github.com/GoogleCloudDataproc/spark-bigquery-connector#properties
// for option information.
val wordsDF =
  spark.read.bigquery("bigquery-public-data:samples.shakespeare")
  .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")
  .save("wordcount_dataset.wordcount_output"))

Ejecuta el código en tu clúster

Usa SSH para conectarte al nodo principal del clúster de Managed Service para Apache Spark
1. Ve a la página Clústeres de Managed Service para Apache Spark en la consola de Google Cloud y, luego, haz clic en el nombre de tu clúster
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 de la instancia principal del clúster>
  Se abrirá una ventana del navegador en tu directorio principal en la instancia principal
```
    Connected, host fingerprint: ssh-rsa 2048 ...
    ...
    user@clusterName-m:~$
    
```
Crea wordcount.scala con el editor de texto vi, vim o nano preinstalado y, luego, pega el código de Scala desde la lista de códigos de Scala.
```
nano wordcount.scala
  
```

Inicia el REPL de spark-shell.

$ spark-shell --jars=gs://spark-lib/bigquery/spark-bigquery-latest_2.12.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>

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 página BigQuery, selecciona la tabla wordcount_output y, luego, haz clic en Vista previa. Tabla de vista previa en la página Explorador de BigQuery en la consola de Cloud.

Tabla de vista previa en la página Explorador de BigQuery en la consola de Cloud.

PySpark

Examina el código y reemplaza el marcador de posición [bucket] por el bucket de Cloud Storage que creaste antes.

#!/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)
# Enable reading from BigQuery views and materializing query results.
spark.conf.set('viewsEnabled', 'true')
spark.conf.set('materializationDataset', 'wordcount_dataset')

# Load data from BigQuery.
words = spark.read.format('bigquery') \
  .load('bigquery-public-data:samples.shakespeare')
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') \
  .save('wordcount_dataset.wordcount_output')

Ejecuta el código en tu clúster
Usa Managed Service para Apache Spark para enviar el código de PySpark: En lugar de ejecutar el código de PySpark de forma manual desde la instancia principal del clúster, puedes enviar el archivo de PySpark directamente a tu clúster (consulta las guías de inicio rápido de Managed Service para Apache Spark). Estos son los pasos para usar Google Cloud CLI:
1. Copia el código de PySpark de la lista de código de PySpark para crear wordcount.py de forma local en un editor de texto
2. Ejecuta el código de PySpark mediante el envío del trabajo a tu clúster con el comando gcloud dataproc jobs submit:
```
gcloud dataproc jobs submit pyspark wordcount.py \
    --cluster=cluster-name \
    --region=region \
    --jars=gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar
```
1. Usa SSH para conectarte a la instancia principal del clúster de Managed Service para Apache Spark
  1. Ve a la página Clústeres de Managed Service para Apache Spark en la consola de Google Cloud y, luego, haz clic en el nombre de tu clúster
  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 de la instancia principal del clúster
    Se abrirá una ventana del navegador en tu directorio principal en la instancia principal.
```
    Connected, host fingerprint: ssh-rsa 2048 ...
    ...
    user@clusterName-m:~$
    
```
2. Crea wordcount.py con el editor de texto vi, vim o nano preinstalado y, luego, pega el código de PySpark desde 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_2.12.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 salida, abre la página BigQuery, selecciona la tabla wordcount_output y, luego, haz clic en Vista previa.

Sugerencias para solucionar problemas

Puedes examinar los registros de trabajos en Cloud Logging y en el explorador de trabajos de BigQuery para solucionar problemas de los trabajos de Spark que usan el conector de BigQuery.

Los registros del controlador de Managed Service para Apache Spark contienen una entrada BigQueryClient con metadatos de BigQuery que incluyen el jobId:
```
ClassNotFoundException INFO BigQueryClient:.. jobId: JobId{project=PROJECT_ID, job=JOB_ID, location=LOCATION}
```
Los trabajos de BigQuery contienen etiquetas Managed Service for Apache Spark_job_id y Managed Service for Apache Spark_job_uuid:
- Registro:
```
protoPayload.serviceData.jobCompletedEvent.job.jobConfiguration.labels.dataproc_job_id="JOB_ID"
protoPayload.serviceData.jobCompletedEvent.job.jobConfiguration.labels.dataproc_job_uuid="JOB_UUID"
protoPayload.serviceData.jobCompletedEvent.job.jobName.jobId="JOB_NAME"
```
- Explorador de trabajos de BigQuery: Haz clic en un ID de trabajo para ver los detalles del trabajo en Etiquetas en Información del trabajo.

¿Qué sigue?

Consulta Almacenamiento de BigQuery y Spark SQL: Python.
Obtén información para crear un archivo de definición de tablas para una fuente de datos externa.
Obtén información para consultar datos particionados de forma externa.
Consulta las sugerencias para ajustar trabajos de Spark.