Utilizza il client Spark Connect serverless

Il client Managed Service for Apache Spark Connect (google-cloud-spark-connect) è una libreria client Python basata sul client open source Apache Spark Connect. Consente alle tue applicazioni, ai tuoi script e ai tuoi notebook locali di eseguire workload PySpark su una sessione Managed Service for Apache Spark remota tramite il protocollo Spark Connect.

Rispetto all'utilizzo diretto del client Apache Spark Connect standard, google-cloud-spark-connect offre i seguenti vantaggi:

  • Ciclo di vita e autenticazione delle sessioni automatizzati: ManagedSparkSession.builder.getOrCreate() esegue automaticamente il provisioning, la configurazione o il riutilizzo di una sessione interattiva remota di Managed Service for Apache Spark e allega le credenziali Google Cloud di aggiornamento. Il codice scritto con ManagedSparkSession viene eseguito senza modifiche anche quando viene inviato come workload batch di Managed Service for Apache Spark, a condizione che il pacchetto google-cloud-spark-connect sia installato nell'ambiente Python del workload batch.
  • Barre di avanzamento in tempo reale e link di monitoraggio: mostra barre di avanzamento in tempo reale nei notebook e nei terminali interattivi durante la creazione della sessione e l'esecuzione dei job Spark (monitoraggio del completamento delle attività nelle varie fasi), insieme a link diretti alla pagina dei dettagli della sessione nella console Google Cloud , alla UI di Spark e alle singole pagine di esecuzione delle query Spark.
  • Extra per i notebook interattivi: se installato con l'extra interactive, l'importazione di google.cloud.managed_spark_connect in un notebook IPython o Jupyter carica automaticamente:
    • explore_dataframe() (con tecnologia google-colabsqlviz) per esplorare, filtrare, ordinare, aggregare e tracciare in modo interattivo i DataFrame PySpark senza scrivere codice di tracciamento.
    • Comandi magici %%sparksql e %sparksql per eseguire query SQL Spark nelle celle del notebook con visualizzazione interattiva dei risultati e associazione di variabili Python o DataFrame.
    • Comando magico %dpip per installare i pacchetti PyPI direttamente sul driver e sugli esecutori della sessione remota senza creare un'immagine container personalizzata.

Come il client Spark Connect open source, google-cloud-spark-connect non richiede un'installazione locale di Spark. Con l'extra [client], installa solo il pacchetto pyspark-client leggero anziché la distribuzione pyspark completa, quindi non è necessario scaricare Spark o installare Java sulla tua macchina per eseguire PySpark su Managed Service for Apache Spark.

Questo documento mostra come installare, configurare e utilizzare il client.

Prima di iniziare

  1. Assicurati di disporre dei ruoli Identity and Access Management che contengono le autorizzazioni necessarie per gestire le sessioni interattive e i profili di runtime (modelli di sessione).

  2. Se esegui il client al di fuori di Google Cloud, fornisci le credenziali di autenticazione. Imposta la variabile di ambiente GOOGLE_APPLICATION_CREDENTIALS sul percorso del file della chiave del account di servizio.

Installa o disinstalla il client

Puoi installare o disinstallare il pacchetto google-cloud-spark-connect utilizzando pip.

Installa

Per i notebook e gli ambienti di sviluppo locali (come VS Code con Google Cloud Data Agent Kit o JupyterLab), installa il client dalla riga di comando in un ambiente virtuale:

pip install 'google-cloud-spark-connect[interactive,client]'

In questo modo viene installato google-cloud-spark-connect, con due extra facoltativi:

  • interactive: installa le estensioni dei notebook interattivi (explore_dataframe(), %%sparksql e %dpip). Può essere omesso per gli script Python autonomi esterni a un notebook.
  • client: installa il pacchetto pyspark-client leggero, che non richiede una JVM locale.

Negli ambienti di notebook gestiti come Colab Enterprise (dove è preinstallato pyspark completo), esegui %pip install con l'extra spark anziché client in una cella di configurazione dedicata nella parte superiore del notebook:

%pip install 'google-cloud-spark-connect[interactive,spark]'

Disinstalla

Per disinstallare il client, esegui questo comando:

pip uninstall google-cloud-spark-connect pyspark-client

Configura il client

Specifica il progetto e la regione per la sessione. Puoi impostare questi valori utilizzando le variabili di ambiente o l'API Builder nel codice.

Variabili di ambiente

Imposta le variabili di 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

Utilizza i metodi .projectId(), .location() e .sessionId().

spark = (
    ManagedSparkSession.builder
    .projectId("my-project")
    .location("us-central1")
    .sessionId("my-spark-session")
    .getOrCreate()
)

Avvia una sessione Spark

Per avviare una sessione Spark, importa ManagedSparkSession nella tua applicazione o nel tuo notebook PySpark, quindi chiama l'API ManagedSparkSession.builder.getOrCreate(). Ti consigliamo di impostare un ID sessione personalizzato con .sessionId() in modo che la riesecuzione del notebook o l'esecuzione di più notebook entro il durata (TTL) della sessione riutilizzi la sessione attiva ed elimini il tempo di avvio della sessione.

  1. Importa la classe ManagedSparkSession. In un blocco note IPython o Jupyter, l'importazione di google.cloud.managed_spark_connect carica automaticamente anche gli extra del blocco note interattivo (explore_dataframe(), %%sparksql e %dpip) se è installato l'extra interactive.

  2. Chiama .sessionId() e .getOrCreate() per avviare o riutilizzare la sessione.

    from google.cloud.managed_spark_connect import ManagedSparkSession
spark = ManagedSparkSession.builder.sessionId("my-spark-session").getOrCreate()

Durante l'avvio di una nuova sessione, il client visualizza una barra di avanzamento in tempo reale e un link alla pagina dei dettagli della sessione nella console Google Cloud (in VS Code, il link apre la sessione in Data Agent Kit). Se esiste già una sessione attiva con l'ID specificato, il client si connette a questa sessione anziché crearne una nuova. In questo caso, il client ignora la configurazione della sessione nel codice. Per applicare le modifiche alla configurazione, termina la sessione esistente (spark.stop(terminate=True)) o utilizza un ID sessione diverso. Se una sessione con l'ID specificato è già terminata, il client elimina automaticamente la sessione precedente e ne crea una nuova con lo stesso ID. Una volta attivata la sessione:

  • La valutazione di spark in una cella del blocco note mostra i link diretti alla pagina dei dettagli della sessione e all'interfaccia utente di Spark.
  • L'esecuzione di azioni Spark (come df.show(), df.collect() o df.toPandas()) mostra una barra di avanzamento delle attività in tempo reale e un link Query Spark alla pagina di esecuzione dell'operazione nell'UI di Spark.

Configura le proprietà Spark

Per configurare le proprietà di Spark, concatena uno o più metodi .config() al builder.

from google.cloud.managed_spark_connect import ManagedSparkSession

spark = (
    ManagedSparkSession.builder
    .sessionId('my-spark-session')
    .config('spark.executor.memory', '48g')
    .config('spark.executor.cores', '8')
    .getOrCreate()
)

Utilizza la configurazione avanzata

Puoi configurare le impostazioni comuni della sessione, come l'ID sessione, la versione di runtime, la subnet, il account di servizio, il template di sessione, il TTL (.ttl()/.ttlSeconds() e .idleTtl()/.idleTtlSeconds()) o le etichette, direttamente utilizzando i metodi del builder:

import datetime
from google.cloud.managed_spark_connect import ManagedSparkSession

spark = (
    ManagedSparkSession.builder
    .projectId('my-project')
    .location('us-central1')
    .sessionId('my-spark-session')
    .runtimeVersion('3.0')
    .subnetwork('SUBNET')
    .idleTtl(datetime.timedelta(minutes=30))
    .getOrCreate()
)

Per una personalizzazione completa della risorsa di sessione sottostante, passa un oggetto google.cloud.dataproc_v1.Session a .sessionConfig():

from google.cloud.managed_spark_connect import ManagedSparkSession
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 = (
    ManagedSparkSession.builder
    .projectId('my-project')
    .location('us-central1')
    .sessionId('my-spark-session')
    .sessionConfig(session_config)
    .getOrCreate()
)

Riutilizza una sessione denominata

L'impostazione di un ID sessione personalizzato con .sessionId() consente di riutilizzare una singola sessione Spark in più notebook o tra i riavvii del kernel entro il TTL della sessione, eliminando la latenza di avvio della sessione e riducendo i costi.

  1. Nel primo notebook, crea una sessione con un ID personalizzato.

    from google.cloud.managed_spark_connect import ManagedSparkSession

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

  2. In un altro notebook (o dopo aver riavviato il kernel), riutilizza la sessione specificando lo stesso ID sessione.

    from google.cloud.managed_spark_connect import ManagedSparkSession

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

Gli ID sessione devono contenere da 4 a 63 caratteri, iniziare con una lettera minuscola e contenere solo lettere minuscole, numeri e trattini. L'ID non può terminare con un trattino. Le sessioni denominate rimangono attive finché non vengono terminate esplicitamente o non raggiungono il TTL configurato. Se una sessione con l'ID specificato è già in uno stato TERMINATED, TERMINATING o FAILED, il client genera un avviso, elimina la sessione precedente e ne crea una nuova con lo stesso ID (l'eliminazione della sessione precedente rimuove i metadati precedenti e la cronologia della UI di Spark da Managed Service for Apache Spark, anche se i log rimangono disponibili in Cloud Logging).

Se esiste già una sessione di attività attiva con l'ID specificato, il client si connette a questa e ignora la configurazione della sessione nel codice. Per applicare le modifiche alla configurazione, termina la sessione esistente o utilizza un ID sessione diverso.

Per terminare una sessione denominata, chiama spark.stop(terminate=True). La chiamata a spark.stop() senza argomenti disconnette il client, ma lascia in esecuzione una sessione denominata. Le sessioni create senza .sessionId() vengono terminate da spark.stop().

Utilizza i comandi magici di Spark SQL

L'extra interactive include la cell magic %%sparksql e la line magic %sparksql per eseguire query SQL Spark nei notebook Jupyter e visualizzare tabelle e grafici interattivi utilizzando google-colabsqlviz (lo stesso widget di visualizzazione utilizzato da explore_dataframe()).

Quando installi google-cloud-spark-connect con l'extra interactive e importi google.cloud.managed_spark_connect in un notebook, l'estensione magica viene caricata automaticamente.

  1. (Facoltativo) Carica esplicitamente l'estensione magica se non hai importato google.cloud.managed_spark_connect.

    %load_ext google.cloud.managed_spark_magics

  2. (Facoltativo) Configura le impostazioni predefinite.

    %config SparkSql.limit=20

  3. Esegui query SQL utilizzando il comando magico per celle %%sparksql (che esegue il rendering di una tabella interattiva e archivia il DataFrame completo in _df per impostazione predefinita) o il comando magico per righe %sparksql (che sopprime la visualizzazione della tabella e restituisce direttamente il DataFrame):

    %%sparksql
SELECT * FROM your_table

    df = %sparksql SELECT * FROM your_table

  4. (Facoltativo) Passa variabili Python e DataFrame in una query. I valori letterali SQL tipizzati possono essere associati a :name, gli identificatori di tabella o colonna a IDENTIFIER(:name) e i DataFrame nello spazio dei nomi del notebook a {df}:

    min_id = 10
df = spark.createDataFrame([(5, 'a'), (15, 'b')], ['id', 'val'])

    %%sparksql
SELECT * FROM {df} WHERE id >= :min_id

Per utilizzare le opzioni avanzate, aggiungi i flag al comando %%sparksql. Ad esempio, per memorizzare nella cache i risultati e creare una vista, esegui questo comando:

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

Sono disponibili le seguenti opzioni:

  • --cache o -c: memorizza nella cache il DataFrame.
  • --eager o -e: calcola immediatamente il risultato utilizzando localCheckpoint(eager=True).
  • --view VIEW o -v VIEW: crea una visualizzazione temporanea.
  • --limit N o -l N: esegue l'override del limite di visualizzazione delle righe predefinito (solo cell magic).
  • --display MODE o -d MODE: imposta la modalità di visualizzazione (colabsqlviz o html; solo cell magic).
  • --quiet o -q: sopprime la visualizzazione della tabella.
  • --explain [MODE]: stampa il piano di query anziché visualizzare i risultati.
  • --session NAME o -s NAME: viene eseguito in base alla SparkSession nella variabile NAME.
  • variable_name: memorizza il risultato in una variabile (solo cell magic; il valore predefinito è _df).

Installa i pacchetti PyPI nella sessione remota

Per utilizzare pacchetti Python di terze parti all'interno di funzioni definite dall'utente (UDF) PySpark senza creare un'immagine container personalizzata o riavviare la sessione, puoi installare i pacchetti PyPI direttamente sul driver e sugli esecutori della sessione remota:

Magie del notebook

In un notebook interattivo con un ManagedSparkSession attivo, utilizza il comando magico %dpip (caricato automaticamente con google.cloud.managed_spark_connect quando viene installato l'extra interactive):

%dpip install scipy

%dpip installa i pacchetti solo nella sessione remota. Per importare un pacchetto nel codice locale al di fuori di Spark, installalo anche nel tuo ambiente Python locale.

API Python

Negli script o nei notebook, passa pypi=True a spark.addArtifacts():

spark.addArtifacts("scipy", pypi=True)

Passaggi successivi