Ejecuta código de PySpark en notebooks de BigQuery Studio

Puedes crear una sesión de Spark en un notebook de BigQuery Studio para usar código de PySpark y analizar y transformar tus datos a mayor escala. El uso de esta integración administrada te permite agregar paquetes de Python de forma dinámica y procesar grandes conjuntos de datos sin salir del entorno de notebook.

Antes de comenzar

Si aún no lo hiciste, crea un proyecto de Google Cloud y un bucket de Cloud Storage.

1. Configura tu proyecto

   Accede a tu cuenta de Google Cloud . Si eres nuevo en Google Cloud, crea una cuenta para evaluar el rendimiento de nuestros productos en situaciones reales. Los clientes nuevos también obtienen $300 en créditos gratuitos para ejecutar, probar y, además, implementar cargas de trabajo.

   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 role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

   Enable the Managed Service for Apache Spark, BigQuery, and Cloud Storage APIs, if any are not already enabled.

   Roles required to enable APIs

   To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

   Enable the APIs

   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 role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

   Enable the Managed Service for Apache Spark, BigQuery, and Cloud Storage APIs, if any are not already enabled.

   Roles required to enable APIs

   To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

   Enable the APIs

2. Crea un bucket de Cloud Storage en tu proyecto si no tienes uno que puedas usar.

3. Configura tu cuaderno

   • Credenciales del notebook: De forma predeterminada, la sesión del notebook usa tus credenciales de usuario. Como alternativa, puede usar las credenciales de la cuenta de servicio de la sesión.
     • Credenciales de usuario: Tu cuenta de usuario debe tener los siguientes roles de Identity and Access Management:
       • Editor de Dataproc (rol roles/dataproc.editor)
       • Usuario de BigQuery Studio (rol roles/bigquery.studioUser)
       • Rol de usuario de cuenta de servicio (roles/iam.serviceAccountUser) en la cuenta de servicio de la sesión. Este rol contiene el permiso iam.serviceAccounts.actAs necesario para usar la identidad de la cuenta de servicio.
     • Credenciales de cuenta de servicio: Si deseas especificar las credenciales de cuenta de servicio en lugar de las credenciales de usuario para tu sesión de notebook, la cuenta de servicio de la sesión debe tener el siguiente rol:
       • Trabajador de Dataproc (rol roles/dataproc.worker)
   • Entorno de ejecución del notebook: Tu notebook usa un entorno de ejecución predeterminado de Gemini Enterprise Agent Platform, a menos que selecciones uno diferente. Si quieres definir tu propio entorno de ejecución, créalo desde la página Entornos de ejecución en la consola de Google Cloud . Ten en cuenta que, cuando uses la biblioteca de NumPy, debes usar la versión 1.26 de NumPy, que es compatible con Spark 3.5, en el entorno de ejecución del notebook.

Precios

Para obtener información sobre los precios, consulta Precios del entorno de ejecución de notebooks de BigQuery.

Abre un notebook de Python de BigQuery Studio

1. En la consola de Google Cloud , ve a la página BigQuery.

   Ir a BigQuery

2. En la barra de pestañas del panel de detalles, haz clic en la flecha junto al signo + y, luego, en Notebook.

Crea una sesión de Spark en un notebook de BigQuery Studio

Puedes usar un notebook de Python de BigQuery Studio para crear una sesión interactiva de Spark Connect. Cada notebook de BigQuery Studio solo puede tener una sesión de Spark activa asociada.

Puedes crear una sesión de Spark en un notebook de Python de BigQuery Studio de las siguientes maneras:

• Configura y crea una sola sesión en el notebook.
• Configura una sesión de Spark en una plantilla de sesión interactiva y, luego, usa la plantilla para configurar y crear una sesión en el notebook. BigQuery proporciona una función Query using Spark que te ayuda a comenzar a programar la sesión con plantilla, como se explica en la pestaña Sesión de Spark con plantilla.

Sesión única

Para crear una sesión de Spark en un notebook nuevo, haz lo siguiente:

1. En la barra de pestañas del panel del editor, haz clic en la flecha desplegable junto al signo + y, luego, haz clic en Notebook.

   

2. Copia y ejecuta el siguiente código en una celda del notebook para configurar y crear una sesión básica de Spark.

from google.cloud.dataproc_spark_connect import DataprocSparkSession
from google.cloud.dataproc_v1 import Session

import pyspark.sql.functions as f

session = Session()

# Create the Spark session.
spark = (
   DataprocSparkSession.builder
     .appName("APP_NAME")
     .dataprocSessionConfig(session)
     .getOrCreate()
)

Reemplaza lo siguiente:

• APP_NAME: Es un nombre opcional para tu sesión.
• Configuración de sesión opcional: Puedes agregar la configuración de la API de Managed Service para Apache Spark Session para personalizar tu sesión. Estos son algunos ejemplos:
  • RuntimeConfig:

    

    • session.runtime_config.properties={spark.property.key1:VALUE_1,...,spark.property.keyN:VALUE_N}
    • session.runtime_config.container_image = path/to/container/image
  • EnvironmentConfig:

    

    • session.environment_config.execution_config.subnetwork_uri = "SUBNET_NAME"
    • session.environment_config.execution_config.ttl = {"seconds": VALUE}
    • session.environment_config.execution_config.service_account = SERVICE_ACCOUNT

Sesión de Spark basada en plantillas

Puedes ingresar y ejecutar el código en una celda del notebook para crear una sesión de Spark basada en una plantilla de sesión existente. Cualquier parámetro de configuración de session que proporciones en el código de tu notebook anulará cualquiera de los mismos parámetros que se establezcan en la plantilla de sesión.

Para comenzar rápidamente, usa la plantilla Query using Spark para completar previamente tu notebook con el código de la plantilla de sesión de Spark:

1. En la barra de pestañas del panel del editor, haz clic en la flecha desplegable junto al signo + y, luego, haz clic en Notebook.

   

2. En Comienza con una plantilla, haz clic en Consulta con Spark y, luego, en Usar plantilla para insertar el código en tu notebook.

   

3. Especifica las variables como se explica en las Notas.
4. Puedes borrar cualquier celda de código de muestra adicional que se haya insertado en el notebook.

from google.cloud.dataproc_spark_connect import DataprocSparkSession
from google.cloud.dataproc_v1 import Session
session = Session()
project_id = "PROJECT_ID"
location = "LOCATION"
# Configure the session with an existing session template.
session_template = "SESSION_TEMPLATE"
session.session_template = f"projects/{project_id}/locations/{location}/sessionTemplates/{session_template}"
# Create the Spark session.
spark = (
   DataprocSparkSession.builder
     .appName("APP_NAME")
     .dataprocSessionConfig(session)
     .getOrCreate()
)

Reemplaza lo siguiente:

• PROJECT_ID: Es el ID del proyecto, que aparece en la sección Información del proyecto del panel de la consola deGoogle Cloud .
• LOCATION: Es la región de Compute Engine en la que se ejecutará tu sesión de notebook. Si no se proporciona, se usará la región de la VM que crea el notebook.

• SESSION_TEMPLATE: Es el nombre de una plantilla de sesión interactiva existente. Los parámetros de configuración de la sesión se obtienen de la plantilla. La plantilla también debe especificar los siguientes parámetros de configuración:

  • Versión del entorno de ejecución 2.3+

  • Tipo de notebook: Spark Connect

    Ejemplo:

    

• APP_NAME: Es un nombre opcional para tu sesión.

Escribe y ejecuta código de PySpark en tu notebook de BigQuery Studio

Después de crear una sesión de Spark en tu notebook, úsala para ejecutar el código de notebook de Spark en el notebook.

Compatibilidad con la API de PySpark de Spark Connect: Tu sesión de notebook de Spark Connect admite la mayoría de las APIs de PySpark, incluidas DataFrame, Functions y Column, pero no admite SparkContext ni RDD ni otras APIs de PySpark. Para obtener más información, consulta Qué se admite en Spark 3.5.

Sugerencia: Puedes consultar la referencia de la API de Spark SQL para saber si Spark Connect admite una API. La documentación de una API compatible contiene el mensaje "Admite Spark Connect":

Escrituras directas en notebooks de Spark Connect: Las sesiones de Spark en un notebook de BigQuery Studio preconfiguran el conector de Spark BigQuery para realizar escrituras de datos DIRECTAS. El método de escritura DIRECT usa la API de BigQuery Storage Write, que escribe datos directamente en BigQuery. El método de escritura INDIRECT, que es el predeterminado para los lotes de Managed Service para Apache Spark, escribe datos en un bucket intermedio de Cloud Storage y, luego, escribe los datos en BigQuery (para obtener más información sobre las escrituras INDIRECT, consulta Lee y escribe datos desde y hacia BigQuery).

APIs específicas de Managed Service para Apache Spark: Managed Service para Apache Spark simplifica la adición de paquetes de PyPI de forma dinámica a tu sesión de Spark extendiendo el método addArtifacts. Puedes especificar la lista en formato version-scheme (similar a pip install). Esto indica al servidor de Spark Connect que instale los paquetes y sus dependencias en todos los nodos del clúster, lo que los hace disponibles para los trabajadores de tus UDF.

Ejemplo en el que se instalan la versión especificada de textdistance y las bibliotecas de random2 compatibles más recientes en el clúster para permitir que las UDF que usan textdistance y random2 se ejecuten en los nodos trabajadores.

spark.addArtifacts("textdistance==4.6.1", "random2", pypi=True)

Ayuda con el código del notebook: El notebook de BigQuery Studio proporciona ayuda con el código cuando mantienes el puntero sobre el nombre de una clase o un método, y proporciona ayuda para completar el código a medida que lo ingresas.

En el siguiente ejemplo, si ingresas DataprocSparkSession y mantienes el puntero sobre el nombre de esta clase, se mostrará la ayuda de documentación y finalización de código.

Sugerencia: Consulta el cliente de Spark Connect de Dataproc en GitHub para obtener información sobre cómo usar los métodos DataprocSparkSession.builder para configurar sesiones de Spark Connect.

Ejemplos de PySpark en notebooks de BigQuery Studio

En esta sección, se proporcionan ejemplos de notebooks de Python de BigQuery Studio con código de PySpark para realizar las siguientes tareas:

• Ejecutar un recuento de palabras en un conjunto de datos público de Shakespeare
• Crea una tabla de Iceberg con metadatos guardados en el catálogo de entornos de ejecución de Lakehouse.

Wordcount

En el siguiente ejemplo de PySpark, se crea una sesión de Spark y, luego, se cuentan las ocurrencias de palabras en un conjunto de datos público de bigquery-public-data.samples.shakespeare.

# Basic wordcount example
from google.cloud.dataproc_spark_connect import DataprocSparkSession
from google.cloud.dataproc_v1 import Session
import pyspark.sql.functions as f
session = Session()

# Create the Spark session.
spark = (
   DataprocSparkSession.builder
     .appName("APP_NAME")
     .dataprocSessionConfig(session)
     .getOrCreate()
)
# Run a wordcount on the public Shakespeare dataset.
df = spark.read.format("bigquery").option("table", "bigquery-public-data.samples.shakespeare").load()
words_df = df.select(f.explode(f.split(f.col("word"), " ")).alias("word"))
word_counts_df = words_df.filter(f.col("word") != "").groupBy("word").agg(f.count("*").alias("count")).orderBy("word")
word_counts_df.show()

Reemplaza lo siguiente:

• APP_NAME: Es un nombre opcional para tu sesión.

Resultado:

En el resultado de la celda, se muestra una muestra del resultado del recuento de palabras. Para ver los detalles de la sesión en la consola de Google Cloud , haz clic en el vínculo Vista de detalles de la sesión interactiva. Se abrirá un panel de detalles de Spark que proporciona pestañas para Registros, Ejecutores, SQL y Trabajos (incluida una cronología de eventos). Para supervisar tu sesión de Spark, haz clic en Ver IU de Spark en la página de detalles de la sesión.

Interactive Session Detail View: LINK
+------------+-----+
|        word|count|
+------------+-----+
|           '|   42|
|       ''All|    1|
|     ''Among|    1|
|       ''And|    1|
|       ''But|    1|
|    ''Gamut'|    1|
|       ''How|    1|
|        ''Lo|    1|
|      ''Look|    1|
|        ''My|    1|
|       ''Now|    1|
|         ''O|    1|
|      ''Od's|    1|
|       ''The|    1|
|       ''Tis|    4|
|      ''When|    1|
|       ''tis|    1|
|      ''twas|    1|
|          'A|   10|
|'ARTEMIDORUS|    1|
+------------+-----+
only showing top 20 rows

Tabla de Iceberg

Ejecutar código de PySpark para crear una tabla de Iceberg con metadatos del catálogo de tiempo de ejecución de Lakehouse

En el siguiente código de ejemplo, se crea un sample_iceberg_table con metadatos de tabla almacenados en el catálogo de tiempo de ejecución de Lakehouse y, luego, se consulta la tabla.

from google.cloud.dataproc_spark_connect import DataprocSparkSession
from google.cloud.dataproc_v1 import Session
# Create the Dataproc Serverless session.
session = Session()
# Set the session configuration for BigLake Metastore with the Iceberg environment.
project_id = "PROJECT_ID"
region = "REGION"
subnet_name = "SUBNET_NAME"
location = "LOCATION"
session.environment_config.execution_config.subnetwork_uri = f"{subnet_name}"
warehouse_dir = "gs://BUCKET/WAREHOUSE_DIRECTORY"
catalog = "CATALOG"
namespace = "NAMESPACE"
session.runtime_config.properties[f"spark.sql.catalog.{catalog}"] = "org.apache.iceberg.spark.SparkCatalog"
session.runtime_config.properties[f"spark.sql.catalog.{catalog}.catalog-impl"] = "org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog"
session.runtime_config.properties[f"spark.sql.catalog.{catalog}.gcp_project"] = f"{project_id}"
session.runtime_config.properties[f"spark.sql.catalog.{catalog}.gcp_location"] = f"{location}"
session.runtime_config.properties[f"spark.sql.catalog.{catalog}.warehouse"] = f"{warehouse_dir}"
# Create the Spark Connect session.
spark = (
   DataprocSparkSession.builder
     .appName("APP_NAME")
     .dataprocSessionConfig(session)
     .getOrCreate()
)
# Create the namespace in BigQuery.
spark.sql(f"USE `{catalog}`;")
spark.sql(f"CREATE NAMESPACE IF NOT EXISTS `{namespace}`;")
spark.sql(f"USE `{namespace}`;")
# Create the Iceberg table.
spark.sql("DROP TABLE IF EXISTS `sample_iceberg_table`");
spark.sql("CREATE TABLE sample_iceberg_table (id int, data string) USING ICEBERG;")
spark.sql("DESCRIBE sample_iceberg_table;")
# Insert table data and query the table.
spark.sql("INSERT INTO sample_iceberg_table VALUES (1, \"first row\");")
# Alter table, then query and display table data and schema.
spark.sql("ALTER TABLE sample_iceberg_table ADD COLUMNS (newDoubleCol double);")
spark.sql("DESCRIBE sample_iceberg_table;")
df = spark.sql("SELECT * FROM sample_iceberg_table")
df.show()
df.printSchema()

Notas:

• PROJECT_ID: Es el ID del proyecto, que aparece en la sección Información del proyecto del panel de la consola deGoogle Cloud .
• REGION y SUBNET_NAME: Especifica la región de Compute Engine y el nombre de una subred en la región de la sesión. Managed Service para Apache Spark habilita el Acceso privado a Google (PGA) en la subred especificada.
• LOCATION: La BigQuery_metastore_config.location y la spark.sql.catalog.{catalog}.gcp_location predeterminadas son US, pero puedes elegir cualquier ubicación de BigQuery compatible.
• BUCKET y WAREHOUSE_DIRECTORY: El bucket y la carpeta de Cloud Storage que se usan para el directorio del almacén de Iceberg.
• CATALOG y NAMESPACE: El nombre y el espacio de nombres del catálogo de Iceberg se combinan para identificar la tabla de Iceberg (catalog.namespace.table_name).
• APP_NAME: Es un nombre opcional para tu sesión.

En el resultado de la celda, se muestra el sample_iceberg_table con la columna agregada y un vínculo a la página Detalles de la sesión interactiva en la consola de Google Cloud . Se abrirá un panel de detalles de Spark que proporciona pestañas para Registros, Ejecutores, SQL y Trabajos (incluida una cronología de eventos). Puedes hacer clic en Ver IU de Spark en la página de detalles de la sesión para supervisar tu sesión de Spark.

Interactive Session Detail View: LINK
+---+---------+------------+
| id|     data|newDoubleCol|
+---+---------+------------+
|  1|first row|        NULL|
+---+---------+------------+

root
 |-- id: integer (nullable = true)
 |-- data: string (nullable = true)
 |-- newDoubleCol: double (nullable = true)

Visualiza los detalles de la tabla en BigQuery

Sigue estos pasos para verificar los detalles de la tabla de Iceberg en BigQuery:

1. En la consola de Google Cloud , ve a la página BigQuery.

   Ir a BigQuery

2. En el panel de recursos del proyecto, haz clic en tu proyecto y, luego, en tu espacio de nombres para enumerar la tabla sample_iceberg_table. Haz clic en la tabla Detalles para ver la información de Configuración de la tabla de catálogo abierto.

   Los formatos de entrada y salida son los formatos de clase estándar de Hadoop InputFormat y OutputFormat que usa Iceberg.

   

Otros ejemplos

Crea un DataFrame de Spark (sdf) a partir de un DataFrame de Pandas (df).

sdf = spark.createDataFrame(df)
sdf.show()

Ejecuta agregaciones en Spark DataFrames.

from pyspark.sql import functions as f

sdf.groupby("segment").agg(
   f.mean("total_spend_per_user").alias("avg_order_value"),
   f.approx_count_distinct("user_id").alias("unique_customers")
).show()

Lee desde BigQuery con el conector de Spark-BigQuery.

spark.conf.set("viewsEnabled","true")
spark.conf.set("materializationDataset","my-bigquery-dataset")

sdf = spark.read.format('bigquery') \
 .load(query)

Escribe código de Spark con Gemini Code Assist

Vista previa

Este producto o función está sujeto a las "Condiciones de la Oferta de la Fase Previa a la DG" en la sección Condiciones Generales del Servicio de las Condiciones Específicas del Servicio. Los productos y las funciones de la fase previa a la DG están disponibles “tal como están” y pueden estar sujetos a asistencia limitada. Para obtener más información, consulta las descripciones de la etapa de lanzamiento.

Puedes pedirle a Gemini Code Assist que genere código de PySpark en tu notebook. Gemini Code Assist recupera y usa tablas relevantes de BigQuery y Dataproc Metastore, y sus esquemas, para generar una respuesta de código.

Para generar código de Gemini Code Assist en tu notebook, haz lo siguiente:

1. Para insertar una celda de código nueva, haz clic en + Código en la barra de herramientas. La nueva celda de código muestra Start coding or generate with AI. Haz clic en Generar.

2. En el editor de Generar, ingresa una instrucción en lenguaje natural y, luego, haz clic en enter. Asegúrate de incluir la palabra clave spark o pyspark en tu instrucción.

   Instrucción de muestra:

   create a spark dataframe from order_items and filter to orders created in 2024
   

   Resultado de muestra:

   spark.read.format("bigquery").option("table", "sqlgen-testing.pysparkeval_ecommerce.order_items").load().filter("year(created_at) = 2024").createOrReplaceTempView("order_items")
   df = spark.sql("SELECT * FROM order_items")
   

Sugerencias para la generación de código con Gemini Code Assist

• Para permitir que Gemini Code Assist recupere tablas y esquemas pertinentes, activa la sincronización de Data Catalog para las instancias de Dataproc Metastore.

• Asegúrate de que tu cuenta de usuario tenga acceso a las tablas de consulta de Data Catalog. Para ello, asigna el rol DataCatalog.Viewer.

Finaliza la sesión de Spark

Puedes realizar cualquiera de las siguientes acciones para detener tu sesión de Spark Connect en tu notebook de BigQuery Studio:

• Ejecuta spark.stop() en una celda del notebook.
• Finaliza el entorno de ejecución en el notebook:
  1. Haz clic en el selector de entorno de ejecución y, luego, en Administrar sesiones.

     

  2. En el cuadro de diálogo Sesiones activas, haz clic en el ícono de finalización y, luego, en Finalizar.

     

Organiza el código de los notebooks de BigQuery Studio

Puedes organizar el código de los notebooks de BigQuery Studio de las siguientes maneras:

• Programa el código del notebook desde la consola de Google Cloud (se aplican los precios de los notebooks).

• Ejecuta el código del notebook como una carga de trabajo por lotes (se aplican los precios de Managed Service para Apache Spark).

Programa el código del notebook desde la consola de Google Cloud

Puedes programar el código del notebook de las siguientes maneras:

• Programa el notebook.
• Si la ejecución del código del notebook forma parte de un flujo de trabajo, programa el notebook como parte de una canalización.

Ejecuta el código del notebook como una carga de trabajo por lotes

Completa los siguientes pasos para ejecutar el código del notebook de BigQuery Studio como una carga de trabajo por lotes.

1. Descarga el código del notebook en un archivo en una terminal local o en Cloud Shell.

   Se recomienda descargar y trabajar en Cloud Shell, ya que preinstala editores de texto y otras herramientas, y proporciona compatibilidad con Python integrada.

   1. En la consola de Google Cloud , en la página BigQuery Studio, abre el notebook en el panel Explorador.

   2. Para expandir la barra de menú, haz clic en keyboard_arrow_down Alternar visibilidad del encabezado.

   3. Haz clic en Archivo > Descargar y, luego, en Download.py.

      

2. Generar requirements.txt.

   1. Instala pipreqs en el directorio en el que guardaste el archivo .py.

      pip install pipreqs
      

   2. Ejecuta pipreqs para generar requirements.txt.

      pipreqs filename.py
      

   3. Usa la Google Cloud CLI para copiar el archivo requirements.txt local en un bucket de Cloud Storage.

      gcloud storage cp requirements.txt gs://BUCKET/
      

3. Actualiza el código de la sesión de Spark editando el archivo .py descargado.

   1. Quita o marca como comentario los comandos de la secuencia de comandos de shell.

   2. Quita el código que configura la sesión de Spark y, luego, especifica los parámetros de configuración como parámetros de envío de cargas de trabajo por lotes. (consulta Envía una carga de trabajo por lotes de Spark).

      Ejemplo:

      • Quita la siguiente línea de configuración de la subred de la sesión del código:

        session.environment_config.execution_config.subnetwork_uri = "{subnet_name}"
        

      • Cuando ejecutes tu carga de trabajo por lotes, usa la marca --subnet para especificar la subred.

        gcloud dataproc batches submit pyspark \
        --subnet=SUBNET_NAME
        

   3. Usa un fragmento de código simple para crear sesiones.

      • Muestra del código del notebook descargado antes de la simplificación.

        from google.cloud.dataproc_spark_connect import DataprocSparkSession
        from google.cloud.dataproc_v1 import Session
        
        session = Session()
        spark = DataprocSparkSession \
            .builder \
            .appName("CustomSparkSession")
            .dataprocSessionConfig(session) \
            .getOrCreate()
        

      • Código de la carga de trabajo de Batch después de la simplificación.

        from pyspark.sql import SparkSession
        
        spark = SparkSession \
        .builder \
        .getOrCreate()
        

4. Ejecuta la carga de trabajo por lotes.

   1. Consulta Envía la carga de trabajo por lotes de Spark para obtener instrucciones.

      • Asegúrate de incluir la marca --deps-bucket para que apunte al bucket de Cloud Storage que contiene tu archivo requirements.txt.

        Ejemplo:

      gcloud dataproc batches submit pyspark FILENAME.py \
          --region=REGION \
          --deps-bucket=BUCKET \
          --version=2.3
      

      Notas:

      • FILENAME: Es el nombre del archivo de código del notebook que descargaste y editaste.
      • REGION: Es la región de Compute Engine en la que se encuentra tu clúster.
      • BUCKET: Es el nombre del bucket de Cloud Storage que contiene tu archivo requirements.txt.
      • --version: Se selecciona la versión 2.3 del entorno de ejecución de Spark para ejecutar la carga de trabajo por lotes.

5. Confirma tu código.

   1. Después de probar el código de tu carga de trabajo por lotes, puedes confirmar el archivo .ipynb o .py en tu repositorio con tu cliente de git, como GitHub, GitLab o Bitbucket, como parte de tu canalización de CI/CD.

6. Programa tu carga de trabajo por lotes con Managed Service para Apache Airflow.

   1. Consulta Ejecuta cargas de trabajo de Managed Service para Apache Spark con Managed Airflow para obtener instrucciones.

Solucionar problemas de notebooks

Si se produce un error en una celda que contiene código de Spark, puedes solucionar problemas el problema haciendo clic en el vínculo Vista de detalles de la sesión interactiva en el resultado de la celda (consulta los ejemplos de tablas de Iceberg y Wordcount).

Cuando encuentras un error de código de notebook, navegar al último trabajo de Spark en la IU de Spark suele proporcionar información adicional para ayudarte a depurar el trabajo con errores.

Problemas conocidos y soluciones

Error: Un entorno de ejecución de notebook creado con la versión 3.10 de Python puede causar un error PYTHON_VERSION_MISMATCH cuando intenta conectarse a la sesión de Spark.

Solución: Vuelve a crear el entorno de ejecución con la versión 3.11 de Python.

¿Qué sigue?

• Demostración en video de YouTube: Aprovecha la potencia de Apache Spark integrado en BigQuery.
• Usa el catálogo de tiempo de ejecución de Lakehouse con Managed Service para Apache Spark
• Usa el catálogo de tiempo de ejecución de Lakehouse con Managed Service para Apache Spark