Eseguire query sui cataloghi federati con Spark e il catalogo di runtime Lakehouse

Anteprima

Questa funzionalità è soggetta ai "Termini dell'offerta pre-GA" nella sezione dei Termini di servizio generali dei Termini specifici dei servizi. Le funzionalità pre-GA sono disponibili "così come sono" e potrebbero avere un supporto limitato. Per ulteriori informazioni, consulta le descrizioni della fase di lancio.

Dopo aver configurato l'accesso ai dati cross-cloud nel catalogo di runtime Lakehouse, puoi eseguire query su tabelle Iceberg remote da Managed Service for Apache Spark senza migrare i file o gestire credenzialspazio di archiviazione sul cloudge separate.

Il catalogo di runtime Lakehouse gestisce la sincronizzazione dei metadati e l'accesso sicuro ai dati tramite la distribuzione delle credenziali. Quando Spark esegue una query su una tabella federata, il catalogo di runtime Lakehouse traduce i metadati del catalogo ed emette credenziali con ambito di breve durata per consentire a Spark di leggere i dati da Amazon S3 o Cloud Storage, trattando le tabelle come se fossero locali nel tuo ambienteGoogle Cloud .

Tipi di cataloghi federati supportati

Managed Service for Apache Spark supporta l'esecuzione di query sui seguenti tipi di cataloghi federati tramite l'endpoint del catalogo REST Iceberg del catalogo di runtime Lakehouse:

Tipo di catalogo federato	Backend di archiviazione supportati	Guida alla configurazione nel catalogo di runtime Lakehouse
AWS Glue Data Catalog	Amazon S3 (s3://: catalogo dell'account AWS predefinito o bucket di tabelle Amazon S3)	Connettiti ad AWS Glue
Databricks Unity Catalog	Posizioni esterne Amazon S3 (s3://) o Cloud Storage (gs://)	Connettiti a Databricks Unity Catalog
Catalogo Snowflake Horizon	Amazon S3 (s3://) o Cloud Storage (gs://)	Connettersi a Snowflake Horizon Catalog

Per informazioni sui requisiti di preparazione del catalogo e sulle definizioni dei parametri Spark sottostanti per ogni provider di archiviazione, consulta Eseguire query sui dati remoti nel catalogo runtime Lakehouse e Parametri di configurazione di Spark.

Prima di iniziare

Accedi al tuo account Google Cloud . Se non conosci Google Cloud, crea un account per valutare le prestazioni dei nostri prodotti in scenari reali. I nuovi clienti ricevono anche 300 $di crediti senza costi per l'esecuzione, il test e il deployment dei workload.

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

Verify that you have the permissions required to complete this guide.

Verify that billing is enabled for your Google Cloud project.

Enable the Dataproc, BigLake, and BigQuery 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

Verify that you have the permissions required to complete this guide.

Verify that billing is enabled for your Google Cloud project.

Enable the Dataproc, BigLake, and BigQuery 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

1. Assicurati di avere tabelle Iceberg nel catalogo remoto.
2. Configura una connessione cross-cloud e un catalogo federato nel catalogo di runtime Lakehouse per il tuo provider di cataloghi remoto:
   • AWS Glue Data Catalog
   • Databricks Unity Catalog
   • Snowflake Horizon Catalog

Ruoli IAM (Identity and Access Management) richiesti

Per eseguire gli esempi in questa pagina sono necessari determinati ruoli IAM. A seconda delle policy dell'organizzazione, questi ruoli potrebbero essere già stati concessi. Per controllare le concessioni dei ruoli, consulta Devi concedere i ruoli?. Per saperne di più sulla concessione dei ruoli, consulta Gestisci l'accesso a progetti, cartelle e organizzazioni.

I ruoli necessari dipendono dal modo in cui viene eseguito il workload. I cluster Managed Service for Apache Spark e i runtime serverless Managed Service for Apache Spark 2.3 e versioni precedenti eseguono i workload come account di servizio. I runtime serverless 3.0 e versioni successive di Managed Service for Apache Spark eseguono i carichi di lavoro con le credenziali dell'utente finale per impostazione predefinita, pertanto concedi i ruoli di accesso ai dati al tuo account utente anziché a un account di servizio. Se specifichi un service account personalizzato per un runtime 3.0 o versioni successive, il workload viene eseguito con le credenziali dell'account di servizio. Per saperne di più, consulta Ruoli dei deployment serverless.

Ruoli utente

Per ottenere le autorizzazioni necessarie per creare un cluster Managed Service for Apache Spark, inviare un workload batch o creare una sessione interattiva, chiedi all'amministratore di concederti i seguenti ruoli IAM:

• Utilizza un cluster o un runtime serverless 2.3 o versioni precedenti:
  • Editor Dataproc (roles/dataproc.editor) sul progetto
  • Utente dell'account di servizio (roles/iam.serviceAccountUser) nel service account predefinito di Compute Engine o in un service account personalizzato
• Utilizza un runtime serverless 3.0 o versioni successive: Dataproc Serverless Editor (roles/dataproc.serverlessEditor) sul progetto
• Leggi le tabelle federate con le credenziali dell'utente finale su un runtime serverless 3.0 o versioni successive: Visualizzatore BigLake (roles/biglake.viewer) sul progetto che contiene il catalogo federato
• Utilizza le credenziali dell'account di servizio su un runtime serverless 3.0 o versioni successive: Utente dell'account (roles/iam.serviceAccountUser) sul service account personalizzato

Ruoli del service account

Concedi i seguenti ruoli al account di servizio che esegue il tuo workload. I runtime serverless 3.0 e versioni successive di Managed Service for Apache Spark richiedono questi ruoli solo se il workload viene eseguito con le credenziali del account di servizio.

Per assicurarti che il account di servizio predefinito di Compute Engine o un account di servizio personalizzato disponga delle autorizzazioni necessarie per eseguire i job di Managed Service for Apache Spark e leggere le tabelle federate, chiedi all'amministratore di concedere i seguenti ruoli IAM al account di servizio predefinito di Compute Engine o a un account di servizio personalizzato:

Importante: devi concedere questi ruoli al account di servizio Compute Engine predefinito o a un account di servizio personalizzato, non al tuo account utente. La mancata concessione dei ruoli all'entità corretta potrebbe comportare errori di autorizzazione.

• Esegui job Managed Service for Apache Spark su un cluster o un runtime serverless 2.3 o versioni precedenti: Dataproc Worker (roles/dataproc.worker) sul progetto
• Esegui i job Managed Service for Apache Spark su un runtime serverless 3.0 o versioni successive:
  • Dataproc Serverless Node (roles/dataproc.serverlessNode) sul progetto
  • Dataproc Resource Manager Node Service Agent (roles/dataprocrm.nodeServiceAgent) sul progetto
• Scopri e leggi i metadati delle tabelle federate in fase di runtime: BigLake Viewer (roles/biglake.viewer) nel progetto che contiene il catalogo federato

Per assicurarti che l'agente di servizio Managed Service for Apache Spark ([e-mail]) disponga delle autorizzazioni necessarie per risolvere la proprietà di configurazione semplificata dataproc.lakehouse.catalog. quando crei un cluster, invii un workload batch o crei una sessione interattiva, chiedi all'amministratore di concedere il ruolo IAM Visualizzatore BigLake (roles/biglake.viewer) all'agente di servizio Managed Service for Apache Spark ([e-mail]) sul progetto che contiene il catalogo federato.

Importante: devi concedere questo ruolo al service agent di Managed Service for Apache Spark ([e-mail]), non al tuo account utente. La mancata concessione del ruolo all'entità corretta potrebbe comportare errori di autorizzazione.

Configura la connettività di rete in uscita per Amazon S3

Quando esegui query su cataloghi federati supportati da Amazon S3, i workload batch serverless e le sessioni interattive di Managed Service for Apache Spark e i cluster Managed Service for Apache Spark con IP interno non possono connettersi agli endpoint Amazon S3 con la loro configurazione di rete predefinita. Esegui il provisioning di un router Cloud e di Cloud NAT nella rete VPC del tuo workload per consentire il traffico in uscita verso Amazon S3:

gcloud compute routers create lakehouse-router \
    --network=NETWORK_NAME \
    --region=REGION

gcloud compute routers nats create lakehouse-nat \
    --router=lakehouse-router \
    --auto-allocate-nat-external-ips \
    --nat-all-subnet-ip-ranges \
    --region=REGION

Sostituisci quanto segue:

• NETWORK_NAME: la rete VPC per il cluster, il workload batch o la sessione interattiva di Managed Service for Apache Spark, ad esempio default.
• REGION: la regione Google Cloud del cluster, del workload batch o della sessione interattiva di Managed Service for Apache Spark.

Configura Managed Service for Apache Spark ed esegui query sulle tabelle federate

Puoi configurare un cluster Managed Service for Apache Spark, un workload batch serverless o una sessione interattiva per connetterti al catalogo federato nel catalogo di runtime Lakehouse. Queste risorse di Managed Service for Apache Spark supportano la proprietà di configurazione semplificata dataproc.lakehouse.catalog.CATALOG_NAME (consigliata), che legge la definizione del catalogo federato dal catalogo di runtime Lakehouse e genera le proprietà Spark del catalogo REST Iceberg richieste, incluse le intestazioni di archiviazione io-impl e di distribuzione delle credenziali (la forma del cluster della proprietà aggiunge il prefisso dataproc:).

Sia la configurazione semplificata che quella manuale utilizzano l'autorizzazione Google per Iceberg (GoogleAuthManager), che richiede una delle seguenti versioni di runtime serverless o immagine cluster:

• Cluster Managed Service for Apache Spark: versioni immagine 2.2 2.2.65 e successive o versioni immagine 2.3 2.3.11 e successive.
• Batch workload serverless e sessioni interattive di Managed Service for Apache Spark: runtime 2.2.60 e versioni successive, runtime 2.3.10 e versioni successive o runtime 3.0.

Per i workload batch serverless e le sessioni interattive, i ruoli richiesti dipendono dalla versione di runtime. I runtime 3.0 e versioni successive utilizzano le credenziali utente finale per impostazione predefinita. Per maggiori dettagli, consulta Ruoli IAM richiesti e Ruoli di deployment serverless.

Cluster

Crea un cluster Managed Service for Apache Spark con il componente facoltativo Iceberg utilizzando le proprietà di configurazione semplificate (consigliato) o specificando manualmente le proprietà del catalogo.

Configurazione semplificata tramite le proprietà (consigliata)

Crea il cluster con la proprietà del catalogo federato:

gcloud dataproc clusters create CLUSTER_NAME \
  --project=PROJECT_ID \
  --region=REGION \
  --image-version=IMAGE_VERSION \
  --optional-components=ICEBERG,JUPYTER \
  --enable-component-gateway \
  --properties="dataproc:dataproc.lakehouse.catalog.CATALOG_NAME=projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,spark:spark.sql.defaultCatalog=CATALOG_NAME"

Sostituisci quanto segue:

• CLUSTER_NAME: un nome per il tuo cluster.
• PROJECT_ID: il tuo ID progettoGoogle Cloud .
• REGION: la regione Google Cloud del cluster, ad esempio us-central1.
• IMAGE_VERSION: la versione dell'immagine di Managed Service for Apache Spark, ad esempio 2.3-debian12. Per le versioni supportate, consulta i requisiti riportati in precedenza in questa sezione.
• CATALOG_NAME: un nome per il catalogo Spark locale, ad esempio federated_catalog. Può essere uguale a FEDERATED_CATALOG_NAME.
• FEDERATED_CATALOG_NAME: l'ID del catalogo federato che hai creato in Prima di iniziare.

Nota: puoi passare più cataloghi separati da virgole al flag --properties. Ogni proprietà del catalogo dei cluster richiede il prefisso dataproc:.

Esecuzione di query su Snowflake Horizon Catalog

Se stai eseguendo una query su un catalogo Snowflake Horizon, tieni presente quanto segue:

• Spazio di archiviazione Snowflake: la configurazione semplificata supporta le tabelle Snowflake archiviate in Amazon S3. Per le tabelle Snowflake archiviate in Cloud Storage, utilizza la configurazione manuale con io-impl impostato su org.apache.iceberg.gcp.gcs.GCSFileIO.

• Colonne Snowflake STRING:se si verificano errori durante l'esecuzione di query sulle colonne Snowflake STRING in cui la serializzazione dell'archiviazione della tabella in Snowflake non è COMPATIBLE, puoi modificare la serializzazione dell'archiviazione in COMPATIBLE in Snowflake o disattivare la vettorizzazione Parquet in Spark aggiungendo le seguenti proprietà al flag --properties:

  spark:spark.sql.iceberg.vectorization.enabled=false,spark:spark.sql.catalog.CATALOG_NAME.table-override.read.parquet.vectorization.enabled=false
      

Nel tuo blocco note Jupyter o nella tua applicazione PySpark, crea SparkSession senza specificare le configurazioni del catalogo:

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("APP_NAME").getOrCreate()

Sostituisci APP_NAME con il nome della tua applicazione Spark.

Configurazione manuale

Se non utilizzi la proprietà di configurazione semplificata, crea il cluster come descritto in precedenza in questa sezione, senza passare la proprietà dataproc:dataproc.lakehouse.catalog. al flag --properties, quindi configura SparkSession manualmente. Per un elenco dei valori dei parametri del fornitore di spazio di archiviazione, consulta Parametri di configurazione di Spark.

from pyspark.sql import SparkSession

catalog_name = "CATALOG_NAME"

spark = SparkSession.builder.appName("APP_NAME") \
.config('spark.sql.defaultCatalog', catalog_name) \
.config(f'spark.sql.catalog.{catalog_name}', 'org.apache.iceberg.spark.SparkCatalog') \
.config(f'spark.sql.catalog.{catalog_name}.type', 'rest') \
.config(f'spark.sql.catalog.{catalog_name}.uri', 'https://biglake.googleapis.com/iceberg/v1/restcatalog') \
.config(f'spark.sql.catalog.{catalog_name}.warehouse', 'bl://projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME') \
.config(f'spark.sql.catalog.{catalog_name}.header.x-goog-user-project', 'PROJECT_ID') \
.config(f'spark.sql.catalog.{catalog_name}.rest.auth.type', 'org.apache.iceberg.gcp.auth.GoogleAuthManager') \
.config(f'spark.sql.catalog.{catalog_name}.header.X-Iceberg-Access-Delegation', 'vended-credentials') \
.config(f'spark.sql.catalog.{catalog_name}.io-impl', 'IO_IMPL') \
.config(f'spark.sql.catalog.{catalog_name}.s3.client-factory-impl', 'org.apache.iceberg.aws.s3.DefaultS3FileIOAwsClientFactory') \
.config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
.getOrCreate()

Sostituisci IO_IMPL con l'implementazione di FileIO per lo spazio di archiviazione che contiene le tue tabelle:

• Tabelle in Amazon S3: org.apache.iceberg.aws.s3.S3FileIO
• Tabelle in Cloud Storage: org.apache.iceberg.gcp.gcs.GCSFileIO

La proprietà s3.client-factory-impl si applica solo alle tabelle in Amazon S3. Ometti questa riga per le tabelle Cloud Storage.

Serverless

Invia un workload batch PySpark a Managed Service for Apache Spark serverless utilizzando le proprietà di configurazione semplificate (consigliato) o specificando manualmente le proprietà del catalogo.

Configurazione semplificata utilizzando le proprietà (opzione consigliata)

Passa la proprietà del catalogo federato senza il prefisso dataproc: quando invii il workload batch:

gcloud dataproc batches submit pyspark PYSPARK_FILE \
  --project=PROJECT_ID \
  --region=REGION \
  --version=RUNTIME_VERSION \
  --properties="dataproc.lakehouse.catalog.CATALOG_NAME=projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,spark.sql.defaultCatalog=CATALOG_NAME"

Sostituisci quanto segue:

• PYSPARK_FILE: il percorso Cloud Storage gs:// del file dell'applicazione PySpark.
• PROJECT_ID: il tuo ID progettoGoogle Cloud .
• REGION: la regione del batch workload.
• RUNTIME_VERSION: la versione del runtime serverless di Managed Service for Apache Spark, ad esempio 3.0. Per le versioni supportate, consulta i requisiti riportati in precedenza in questa sezione. I ruoli IAM richiesti dipendono dalla versione del runtime.
• CATALOG_NAME: un nome per il catalogo Spark locale, ad esempio federated_catalog. Può essere uguale a FEDERATED_CATALOG_NAME.
• FEDERATED_CATALOG_NAME: l'ID del catalogo federato che hai creato in Prima di iniziare.

Nota: puoi passare più cataloghi separati da virgole al flag --properties.

Esecuzione di query su Snowflake Horizon Catalog

Se stai eseguendo una query su un catalogo Snowflake Horizon, tieni presente quanto segue:

• Spazio di archiviazione Snowflake: la configurazione semplificata supporta le tabelle Snowflake archiviate in Amazon S3. Per le tabelle Snowflake archiviate in Cloud Storage, utilizza la configurazione manuale con io-impl impostato su org.apache.iceberg.gcp.gcs.GCSFileIO.

• Colonne Snowflake STRING: se si verificano errori durante l'esecuzione di query sulle colonne Snowflake STRING in cui la serializzazione dell'archiviazione della tabella in Snowflake non è COMPATIBLE, puoi modificare la serializzazione dell'archiviazione in COMPATIBLE in Snowflake o disattivare la vettorizzazione Parquet in Spark aggiungendo le seguenti proprietà al flag --properties:

  spark.sql.iceberg.vectorization.enabled=false,spark.sql.catalog.CATALOG_NAME.table-override.read.parquet.vectorization.enabled=false
      

Nel file dell'applicazione PySpark, crea SparkSession senza specificare le configurazioni del catalogo:

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("APP_NAME").getOrCreate()

Configurazione manuale

Se non utilizzi la proprietà di configurazione semplificata, specifica le proprietà del catalogo con il flag --properties del comando gcloud dataproc batches submit. Per un elenco dei valori dei parametri del fornitore di spazio di archiviazione, consulta Parametri di configurazione di Spark.

gcloud dataproc batches submit pyspark PYSPARK_FILE \
  --project=PROJECT_ID \
  --region=REGION \
  --version=RUNTIME_VERSION \
  --properties="\
spark.sql.defaultCatalog=CATALOG_NAME,\
spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog,\
spark.sql.catalog.CATALOG_NAME.type=rest,\
spark.sql.catalog.CATALOG_NAME.uri=https://biglake.googleapis.com/iceberg/v1/restcatalog,\
spark.sql.catalog.CATALOG_NAME.warehouse=bl://projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,\
spark.sql.catalog.CATALOG_NAME.header.x-goog-user-project=PROJECT_ID,\
spark.sql.catalog.CATALOG_NAME.rest.auth.type=org.apache.iceberg.gcp.auth.GoogleAuthManager,\
spark.sql.catalog.CATALOG_NAME.header.X-Iceberg-Access-Delegation=vended-credentials,\
spark.sql.catalog.CATALOG_NAME.io-impl=IO_IMPL,\
spark.sql.catalog.CATALOG_NAME.s3.client-factory-impl=org.apache.iceberg.aws.s3.DefaultS3FileIOAwsClientFactory,\
spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions"

Sostituisci IO_IMPL con l'implementazione di FileIO per lo spazio di archiviazione che contiene le tue tabelle:

• Tabelle in Amazon S3: org.apache.iceberg.aws.s3.S3FileIO
• Tabelle in Cloud Storage: org.apache.iceberg.gcp.gcs.GCSFileIO

La proprietà s3.client-factory-impl si applica solo alle tabelle in Amazon S3. Ometti questa proprietà per le tabelle Cloud Storage.

Serverless

Crea una sessione interattiva serverless Managed Service for Apache Spark in un notebook tramite l'API DataprocSparkSession utilizzando proprietà di configurazione semplificate (consigliato) o specificando manualmente le proprietà del catalogo.

Configurazione semplificata tramite le proprietà (consigliata)

In un notebook in cui è installato il pacchetto dataproc-spark-connect, ad esempio un notebook BigQuery Studio o un notebook Vertex AI Workbench, passa la proprietà del catalogo federato senza il prefisso dataproc: quando crei la sessione:

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

catalog_name = "CATALOG_NAME"

session_config = Session()
session_config.runtime_config.version = 'RUNTIME_VERSION'

spark = DataprocSparkSession.builder.appName("APP_NAME") \
.dataprocSessionConfig(session_config) \
.config(f'dataproc.lakehouse.catalog.{catalog_name}', 'projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME') \
.config('spark.sql.defaultCatalog', catalog_name) \
.getOrCreate()

Sostituisci quanto segue:

• CATALOG_NAME: un nome per il catalogo Spark locale, ad esempio federated_catalog. Può essere uguale a FEDERATED_CATALOG_NAME.
• RUNTIME_VERSION: la versione del runtime serverless di Managed Service for Apache Spark, ad esempio 3.0. Per le versioni supportate, consulta i requisiti riportati in precedenza in questa sezione. I ruoli IAM richiesti dipendono dalla versione del runtime.
• APP_NAME: il nome dell'applicazione Spark.
• PROJECT_ID: il tuo Google Cloud ID progetto.
• FEDERATED_CATALOG_NAME: l'ID del catalogo federato che hai creato in Prima di iniziare.

Puoi anche impostare queste proprietà in un template di sessione.

Esecuzione di query su Snowflake Horizon Catalog

Se stai eseguendo una query su un catalogo Snowflake Horizon, tieni presente quanto segue:

• Spazio di archiviazione Snowflake: la configurazione semplificata supporta le tabelle Snowflake archiviate in Amazon S3. Per le tabelle Snowflake archiviate in Cloud Storage, utilizza la configurazione manuale con io-impl impostato su org.apache.iceberg.gcp.gcs.GCSFileIO.

• Colonne Snowflake STRING: se riscontri errori durante l'esecuzione di query sulle colonne Snowflake STRING in cui la policy di serializzazione dello spazio di archiviazione della tabella in Snowflake non è COMPATIBLE, puoi modificare la policy di serializzazione dello spazio di archiviazione in COMPATIBLE in Snowflake o disattivare la vettorizzazione Parquet in Spark aggiungendo le seguenti chiamate .config() al generatore di sessioni:

  .config('spark.sql.iceberg.vectorization.enabled', 'false') \
  .config(f'spark.sql.catalog.{catalog_name}.table-override.read.parquet.vectorization.enabled', 'false')
      

Configurazione manuale

Se non utilizzi la proprietà di configurazione semplificata, specifica le proprietà del catalogo quando crei la sessione. Per un elenco dei valori dei parametri del fornitore di spazio di archiviazione, consulta Parametri di configurazione di Spark.

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

catalog_name = "CATALOG_NAME"

session_config = Session()
session_config.runtime_config.version = 'RUNTIME_VERSION'

spark = DataprocSparkSession.builder.appName("APP_NAME") \
.dataprocSessionConfig(session_config) \
.config('spark.sql.defaultCatalog', catalog_name) \
.config(f'spark.sql.catalog.{catalog_name}', 'org.apache.iceberg.spark.SparkCatalog') \
.config(f'spark.sql.catalog.{catalog_name}.type', 'rest') \
.config(f'spark.sql.catalog.{catalog_name}.uri', 'https://biglake.googleapis.com/iceberg/v1/restcatalog') \
.config(f'spark.sql.catalog.{catalog_name}.warehouse', 'bl://projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME') \
.config(f'spark.sql.catalog.{catalog_name}.header.x-goog-user-project', 'PROJECT_ID') \
.config(f'spark.sql.catalog.{catalog_name}.rest.auth.type', 'org.apache.iceberg.gcp.auth.GoogleAuthManager') \
.config(f'spark.sql.catalog.{catalog_name}.header.X-Iceberg-Access-Delegation', 'vended-credentials') \
.config(f'spark.sql.catalog.{catalog_name}.io-impl', 'IO_IMPL') \
.config(f'spark.sql.catalog.{catalog_name}.s3.client-factory-impl', 'org.apache.iceberg.aws.s3.DefaultS3FileIOAwsClientFactory') \
.config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
.getOrCreate()

Sostituisci RUNTIME_VERSION come descritto nell'esempio di configurazione semplificata riportato in precedenza in questa scheda.

Sostituisci IO_IMPL con l'implementazione FileIO per lo spazio di archiviazione che contiene le tue tabelle:

• Tabelle in Amazon S3: org.apache.iceberg.aws.s3.S3FileIO
• Tabelle in Cloud Storage: org.apache.iceberg.gcp.gcs.GCSFileIO

La proprietà s3.client-factory-impl si applica solo alle tabelle in Amazon S3. Ometti questa riga per le tabelle Cloud Storage.

Esegui query su tabelle federate

Dopo aver configurato la sessione Spark, puoi eseguire query sulle tabelle nel catalogo federato utilizzando PySpark o Spark SQL.

Nota: la proprietà dataproc.lakehouse.catalog. registra il catalogo con Spark, ma non lo imposta come catalogo predefinito. Gli esempi precedenti impostano la proprietà del cluster spark.sql.defaultCatalog serverless o spark:spark.sql.defaultCatalog, che ti consente di utilizzare nomi di tabelle non qualificati. Se ometti queste proprietà, devi qualificare i nomi delle tabelle con il nome del catalogo, come mostrato negli esempi seguenti.

1. Leggi una tabella federata utilizzando l'API PySpark DataFrame:

   df = spark.table("CATALOG_NAME.NAMESPACE_NAME.TABLE_NAME")
   df.show(10, truncate=False)

   Sostituisci quanto segue:

   • CATALOG_NAME: il nome del catalogo Spark locale. Se imposti spark.sql.defaultCatalog su CATALOG_NAME, puoi omettere questo prefisso.
   • NAMESPACE_NAME: lo spazio dei nomi (database o schema) nel catalogo federato.
   • TABLE_NAME: il nome della tabella nel catalogo federato.

2. Esegui una query sulla tabella federata utilizzando Spark SQL:

   spark.sql("""
       SELECT *
       FROM CATALOG_NAME.NAMESPACE_NAME.TABLE_NAME
       LIMIT 10
   """).show()

Unire o materializzare i dati tra cataloghi federati e locali

I cataloghi federati sono di sola lettura. Per unire i dati cross-cloud remoti con le tabelle locali o materializzare i risultati delle query federate in Cloud Storage, collega il catalogo federato e un catalogo di lettura e scrittura nel catalogo di runtime di Lakehouse utilizzando le proprietà dataproc.lakehouse.catalog. separate da virgole:

Cluster

Esempio di cluster Managed Service for Apache Spark:

--properties="dataproc:dataproc.lakehouse.catalog.FEDERATED_SPARK_CATALOG=projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,dataproc:dataproc.lakehouse.catalog.LOCAL_SPARK_CATALOG=projects/PROJECT_ID/catalogs/LOCAL_CATALOG_ID,spark:spark.sql.defaultCatalog=LOCAL_SPARK_CATALOG"

Serverless

Esempio di workload batch serverless di Managed Service for Apache Spark:

--properties="dataproc.lakehouse.catalog.FEDERATED_SPARK_CATALOG=projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,dataproc.lakehouse.catalog.LOCAL_SPARK_CATALOG=projects/PROJECT_ID/catalogs/LOCAL_CATALOG_ID,spark.sql.defaultCatalog=LOCAL_SPARK_CATALOG"

Serverless

Esempio di sessione interattiva serverless di Managed Service for Apache Spark:

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

session_config = Session()
session_config.runtime_config.version = 'RUNTIME_VERSION'

spark = DataprocSparkSession.builder.appName("APP_NAME") \
  .dataprocSessionConfig(session_config) \
  .config('dataproc.lakehouse.catalog.FEDERATED_SPARK_CATALOG', 'projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME') \
  .config('dataproc.lakehouse.catalog.LOCAL_SPARK_CATALOG', 'projects/PROJECT_ID/catalogs/LOCAL_CATALOG_ID') \
  .config('spark.sql.defaultCatalog', 'LOCAL_SPARK_CATALOG') \
  .getOrCreate()

Sostituisci RUNTIME_VERSION come descritto nell'esempio di sessione interattiva riportato in precedenza in questa pagina.

Dopo aver impostato queste proprietà, puoi eseguire query sul catalogo federato e scrivere i risultati in una tabella nel catalogo di runtime Lakehouse locale.

L'entità che esegue il workload è il service account per i cluster, i runtime serverless 2.3 e versioni precedenti e i runtime serverless 3.0 e versioni successive con credenziali dell'account di servizio. Per i runtime serverless 3.0 e versioni successive con credenziali utente finale (impostazione predefinita), il principal è il tuo account utente.

Per assicurarti che l'entità che esegue il workload disponga delle autorizzazioni necessarie per scrivere i dati della tabella nel catalogo locale, chiedi all'amministratore di concedere i seguenti ruoli IAM all'entità che esegue il workload:

• Scrittura dei dati della tabella in modalità di distribuzione delle credenziali: BigLake Editor (roles/biglake.editor) sul progetto che contiene il catalogo locale
• Scrivi i dati della tabella in modalità di distribuzione non delle credenziali:
  • BigLake Editor (roles/biglake.editor) sul progetto che contiene il catalogo locale
  • Storage Object User (roles/storage.objectUser) su tutti i bucket Cloud Storage associati al catalogo locale

L'esempio seguente crea una tabella nel catalogo locale da una tabella federata:

spark.sql("""
    CREATE TABLE IF NOT EXISTS LOCAL_SPARK_CATALOG.LOCAL_NAMESPACE.materialized_table
    USING iceberg
    AS SELECT *
    FROM FEDERATED_SPARK_CATALOG.REMOTE_NAMESPACE.REMOTE_TABLE
""")

Passaggi successivi

• Crea un lakehouse con Spark e il catalogo di runtime Lakehouse.
• Scopri di più sull'esecuzione di query sui dati remoti nel catalogo di runtime Lakehouse.
• Leggi Informazioni sull'accesso ai dati cross-cloud.
• Scopri di più sulla distribuzione delle credenziali.