Usa el cliente de Spark Connect sin servidores

El cliente de Spark Connect es un wrapper del cliente de Managed Service para Apache Spark Connect. Permite que las aplicaciones se comuniquen con una sesión remota de Managed Service para Apache Spark con el protocolo de Spark Connect. En este documento, se muestra cómo instalar, configurar y usar el cliente.

Antes de comenzar

  1. Asegúrate de tener los roles de Identity and Access Management que contienen los permisos necesarios para administrar las sesiones interactivas y las plantillas de sesión.

  2. Si ejecutas el cliente fuera de Google Cloud, proporciona credenciales de autenticación. Configura la variable de entorno GOOGLE_APPLICATION_CREDENTIALS en la ruta de acceso del archivo de claves de tu cuenta de servicio.

Instala o desinstala el cliente

Puedes instalar o desinstalar el paquete dataproc-spark-connect con pip.

Instalar

Para instalar la versión más reciente del cliente, ejecuta el siguiente comando:

pip install -U dataproc-spark-connect

Desinstalar

Para desinstalar el cliente, ejecuta el siguiente comando:

pip uninstall dataproc-spark-connect

Configura el cliente

Especifica el proyecto y la región de tu sesión. Puedes establecer estos valores con variables de entorno o con la API de Builder en tu código.

Variables de entorno

Configura las variables de entorno GOOGLE_CLOUD_PROJECT y GOOGLE_CLOUD_REGION. 

# Google Cloud Configuration for Managed Spark Connect Integration Tests
# Copy this file to .env and fill in your actual values

# ============================================================================
# REQUIRED CONFIGURATION
# ============================================================================

# Your Google Cloud Project ID
GOOGLE_CLOUD_PROJECT="your-project-id"

# Google Cloud Region where Managed Spark sessions will be created
# Integration tests also accept a comma-separated ordered list of fallback regions
# (e.g. "us-east1,us-central1,us-west1") when using a short subnet name.
GOOGLE_CLOUD_REGION="us-central1"

# Path to service account key file (if using SERVICE_ACCOUNT auth)
GOOGLE_APPLICATION_CREDENTIALS="/path/to/your/service-account-key.json"

# ============================================================================
# AUTHENTICATION CONFIGURATION
# ============================================================================

# Authentication type (SERVICE_ACCOUNT or END_USER_CREDENTIALS). If not set, API default is used.
# MANAGED_SPARK_CONNECT_AUTH_TYPE="SERVICE_ACCOUNT"
# MANAGED_SPARK_CONNECT_AUTH_TYPE="END_USER_CREDENTIALS"

# Service account email for workload authentication (optional)
# MANAGED_SPARK_CONNECT_SERVICE_ACCOUNT="your-service-account@your-project.iam.gserviceaccount.com"

# ============================================================================
# SESSION CONFIGURATION
# ============================================================================

# Session timeout in seconds (how long session stays active)
# MANAGED_SPARK_CONNECT_TTL_SECONDS="3600"

# Session idle timeout in seconds (how long session stays active when idle)
# MANAGED_SPARK_CONNECT_IDLE_TTL_SECONDS="900"

# Automatically terminate session when Python process exits (true/false)
# MANAGED_SPARK_CONNECT_SESSION_TERMINATE_AT_EXIT="false"

# Custom file path for storing active session information
# MANAGED_SPARK_CONNECT_ACTIVE_SESSION_FILE_PATH="/tmp/managed_spark_connect_session"

# ============================================================================
# DATA SOURCE CONFIGURATION
# ============================================================================

# Default data source for Spark SQL (currently only supports "bigquery")
# Only available for Managed Spark runtime version 2.3
# MANAGED_SPARK_CONNECT_DEFAULT_DATASOURCE="bigquery"

# ============================================================================
# ADVANCED CONFIGURATION
# ============================================================================

# Custom Dataproc API endpoint (uncomment if needed)
# GOOGLE_CLOUD_DATAPROC_API_ENDPOINT="your-region-dataproc.googleapis.com"

# Subnet name or URI for Managed Spark Connect (optional; defaults to "default" in tests)
# Accepts a short subnet name (e.g. "default") or full resource URI:
# MANAGED_SPARK_CONNECT_SUBNET="default"
# MANAGED_SPARK_CONNECT_SUBNET="projects/your-project-id/regions/us-central1/subnetworks/your-subnet-name"

API de Builder

Usa los métodos .projectId() y .location().

spark = DataprocSparkSession.builder.projectId("my-project").location("us-central1").getOrCreate()

Inicia una sesión de Spark

Para iniciar una sesión de Spark, agrega las importaciones necesarias a tu aplicación o notebook de PySpark y, luego, llama a la API de DataprocSparkSession.builder.getOrCreate().

  1. Importa la clase DataprocSparkSession.

  2. Llama al método getOrCreate() para iniciar la sesión.

    from google.cloud.dataproc_spark_connect import DataprocSparkSession
spark = DataprocSparkSession.builder.getOrCreate()

Configura las propiedades de Spark

Para configurar las propiedades de Spark, encadena uno o más métodos .config() al compilador.

from google.cloud.dataproc_spark_connect import DataprocSparkSession

spark = DataprocSparkSession.builder.config('spark.executor.memory', '48g').config('spark.executor.cores', '8').getOrCreate()

Usa la configuración avanzada

Para la configuración avanzada, usa la clase Session para personalizar parámetros de configuración, como la subred o la versión del entorno de ejecución.

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

session_config = Session()
session_config.environment_config.execution_config.subnetwork_uri = 'SUBNET'
session_config.runtime_config.version = '3.0'

spark = DataprocSparkSession.builder.projectId('my-project').location('us-central1').dataprocSessionConfig(session_config).getOrCreate()

Reutiliza una sesión con nombre

Las sesiones con nombre te permiten compartir una sola sesión de Spark en varios notebooks y, al mismo tiempo, evitar demoras repetidas en el tiempo de inicio de la sesión.

  1. En tu primer cuaderno, crea una sesión con un ID personalizado.

    from google.cloud.dataproc_spark_connect import DataprocSparkSession

session_id = 'my-ml-pipeline-session'
spark = DataprocSparkSession.builder.dataprocSessionId(session_id).getOrCreate()
df = spark.createDataFrame([(1, 'data')], ['id', 'value'])
df.show()

  2. En otro notebook, reutiliza la sesión especificando el mismo ID de sesión.

    from google.cloud.dataproc_spark_connect import DataprocSparkSession

session_id = 'my-ml-pipeline-session'
spark = DataprocSparkSession.builder.dataprocSessionId(session_id).getOrCreate()
df = spark.createDataFrame([(2, 'more-data')], ['id', 'value'])
df.show()

Los IDs de sesión deben tener entre 4 y 63 caracteres, comenzar con una letra minúscula y contener solo letras minúsculas, números y guiones. El ID no puede terminar con un guion. No se puede volver a usar una sesión con un ID que se encuentre en un estado TERMINATED.

Usa comandos mágicos de Spark SQL

El paquete admite la biblioteca sparksql-magic para ejecutar consultas en SQL de Spark en notebooks de Jupyter. Los comandos mágicos son una función opcional.

  1. Instala las dependencias requeridas.

    pip install IPython sparksql-magic

  2. Carga la extensión mágica.

    %load_ext sparksql_magic

  3. Opcional: Establece la configuración predeterminada.

    %config SparkSql.limit=20

  4. Ejecutar consultas en SQL

    %%sparksql
SELECT * FROM your_table

Para usar las opciones avanzadas, agrega marcas al comando %%sparksql. Por ejemplo, para almacenar en caché los resultados y crear una vista, ejecuta el siguiente comando:

%%sparksql --cache --view result_view df
SELECT * FROM your_table WHERE condition = true

Las siguientes opciones están disponibles:

  • --cache o -c: almacena en caché el DataFrame.
  • --eager o -e: cachés con carga inmediata.
  • --view VIEW o -v VIEW: Crea una vista temporal.
  • --limit N o -l N: anula el límite predeterminado de visualización de filas.
  • variable_name: Almacena el resultado en una variable.

¿Qué sigue?