Usar o cliente do Spark Connect sem servidor

O cliente do Spark Connect é um wrapper do cliente do Serviço Gerenciado para Apache Spark Connect. Ele permite que os aplicativos se comuniquem com uma sessão remota do Serviço Gerenciado para Apache Spark usando o protocolo Spark Connect. Neste documento, mostramos como instalar, configurar e usar o cliente.

Antes de começar

  1. Verifique se você tem os papéis do Identity and Access Management que contêm as permissões necessárias para gerenciar sessões interativas e modelos de sessão.

  2. Se você executar o cliente fora do Google Cloud, forneça credenciais de autenticação. Defina a variável de ambiente GOOGLE_APPLICATION_CREDENTIALS como o caminho do arquivo de chave da conta de serviço.

Instalar ou desinstalar o cliente

É possível instalar ou desinstalar o pacote dataproc-spark-connect usando pip.

Instalar

Para instalar a versão mais recente do cliente, execute o seguinte comando:

pip install -U dataproc-spark-connect

Desinstalar

Para desinstalar o cliente, execute o seguinte comando:

pip uninstall dataproc-spark-connect

Configurar o cliente

Especifique o projeto e a região da sessão. É possível definir esses valores usando variáveis de ambiente ou a API Builder no seu código.

Variáveis de ambiente

Defina as variáveis de ambiente GOOGLE_CLOUD_PROJECT e 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 Builder

Use os métodos .projectId() e .location().

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

Iniciar uma sessão do Spark

Para iniciar uma sessão do Spark, adicione as importações necessárias ao aplicativo ou notebook do PySpark e chame a API DataprocSparkSession.builder.getOrCreate().

  1. Importe a classe DataprocSparkSession.

  2. Chame o método getOrCreate() para iniciar a sessão.

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

Configurar propriedades do Spark

Para configurar as propriedades do Spark, encadeie um ou mais métodos .config() ao builder.

from google.cloud.dataproc_spark_connect import DataprocSparkSession

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

Usar configuração avançada

Para uma configuração avançada, use a classe Session para personalizar configurações como a sub-rede ou a versão do ambiente de execução.

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()

Reutilizar uma sessão nomeada

As sessões nomeadas permitem compartilhar uma única sessão do Spark em vários notebooks, evitando atrasos repetidos no tempo de inicialização da sessão.

  1. No primeiro notebook, crie uma sessão com um 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. Em outro notebook, reutilize a sessão especificando o mesmo ID.

    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()

Os IDs de sessão precisam ter de 4 a 63 caracteres, começar com uma letra minúscula e conter apenas letras minúsculas, números e hifens. O ID não pode terminar com um hífen. Uma sessão com um ID que está em um estado TERMINATED não pode ser reutilizada.

Usar comandos mágicos do Spark SQL

O pacote é compatível com a biblioteca sparksql-magic para executar consultas SQL do Spark em notebooks do Jupyter. Os comandos mágicos são um recurso opcional.

  1. Instale as dependências necessárias.

    pip install IPython sparksql-magic

  2. Carregue a extensão mágica.

    %load_ext sparksql_magic

  3. Opcional: defina as configurações padrão.

    %config SparkSql.limit=20

  4. Executar consultas SQL.

    %%sparksql
SELECT * FROM your_table

Para usar as opções avançadas, adicione flags ao comando %%sparksql. Por exemplo, para armazenar em cache os resultados e criar uma visualização, execute o seguinte comando:

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

As seguintes opções estão disponíveis:

  • --cache ou -c: armazena o DataFrame em cache.
  • --eager ou -e: caches com carregamento antecipado.
  • --view VIEW ou -v VIEW: cria uma visualização temporária.
  • --limit N ou -l N: substitui o limite padrão de exibição de linhas.
  • variable_name: armazena o resultado em uma variável.

A seguir