Configura Spark con le tabelle Delta

Questo documento spiega come configurare Apache Spark con il catalogo runtime Lakehouse per utilizzare le tabelle Delta Lake. Scopri come creare un catalogo Apache Hive, configurare le sessioni Spark per connetterti al metastore ed eseguire workload per creare e registrare tabelle Delta da interrogare in BigQuery.

Nota: non puoi creare una nuova tabella Delta direttamente tramite l' endpoint Hive del catalogo di runtime Lakehouse o utilizzando Hive DDL. Utilizza invece un motore compatibile con Delta, come Apache Spark, per creare i file della tabella sottostante in Cloud Storage e registrare la tabella nel metastore.

Prima di iniziare

1. Leggi Informazioni sull'endpoint del catalogo Hive.
2. Esamina i formati di archiviazione e i tipi di dati supportati.
3. Esamina le limitazioni e le considerazioni.
Accedi al tuo Google Cloud account. 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 carichi di lavoro.

Verify that billing is enabled for your Google Cloud project.

Enable the Lakehouse, Managed Service for Apache Spark 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

Verify that billing is enabled for your Google Cloud project.

Enable the Lakehouse, Managed Service for Apache Spark 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

Ruoli obbligatori

Per ottenere le autorizzazioni necessarie per utilizzare il catalogo di runtime Lakehouse con le tabelle Delta, chiedi all'amministratore di concederti i seguenti ruoli IAM sul tuo progetto:

• Crea un cluster Managed Service for Apache Spark: Editor Dataproc (roles/dataproc.editor)
• Account di servizio del cluster:
  • Worker Dataproc (roles/dataproc.worker)
  • Utente oggetti Storage (roles/storage.objectUser)
  • Editor BigLake (roles/biglake.editor)
  • Service Usage Consumer (roles/serviceusage.serviceUsageConsumer)
• Accesso in scrittura a tutte le risorse del catalogo di runtime Lakehouse: Editor BigLake (roles/biglake.editor)
• Accesso in sola lettura a tutte le risorse del catalogo di runtime Lakehouse: Visualizzatore BigLake (roles/biglake.viewer)

Per saperne di più sulla concessione dei ruoli, consulta Gestisci l'accesso a progetti, cartelle e organizzazioni.

Potresti anche riuscire a ottenere le autorizzazioni richieste tramite i ruoli personalizzati o altri ruoli predefiniti.

Per le istruzioni, consulta Concedere un singolo ruolo.

Workflow generale

Per utilizzare il catalogo di runtime Lakehouse con le tabelle Delta, segui questo workflow:

1. Crea un catalogo Hive Lakehouse.
2. Configura la sessione Spark con il client metastore Lakehouse e le dipendenze di Delta Lake.
3. Crea e popola le tabelle Delta in Cloud Storage utilizzando Spark, che le registra nel metastore.
4. Esegui query sulle tabelle Delta da Apache Spark (lettura e scrittura) o direttamente da BigQuery (sola lettura).

Crea un catalogo Hive Lakehouse

Per utilizzare il catalogo di runtime Lakehouse con Spark e le tabelle Delta, crea un catalogo Hive.

Un catalogo Hive Lakehouse è una raccolta di database Hive. Prima di eseguire i job Spark, crea un catalogo per registrarlo con Lakehouse Metastore. Il catalogo richiede un nome e una località Cloud Storage per i dati.

Console

1. Nella Google Cloud console, apri la pagina Lakehouse.

   Vai a Lakehouse

2. Fai clic su Crea catalogo.

3. Seleziona Catalogo di runtime Lakehouse.

4. In Tipo di catalogo, seleziona Hive Metastore.

5. Nel campo Seleziona un bucket Cloud Storage, inserisci il nome del bucket Cloud Storage per il catalogo. In alternativa, fai clic su Sfoglia per scegliere o creare un bucket.

6. In ID catalogo, inserisci un nome per il catalogo Hive Lakehouse.

7. In Località principale, specifica la stessa regione del bucket.

8. Fai clic su Crea.

gcloud

Per creare un catalogo Hive, esegui questo comando:

gcloud biglake hive catalogs create LAKEHOUSE_CATALOG_ID \
    --project=PROJECT_ID \
    --location-uri="gs://GCS_WAREHOUSE_PATH" \
    --primary-location=REGION \
    --description="DESCRIPTION"

Sostituisci quanto segue:

• LAKEHOUSE_CATALOG_ID: il nome del catalogo Hive.

• GCS_WAREHOUSE_PATH: il percorso Cloud Storage per i dati del data warehouse.

• PROJECT_ID: il tuo Google Cloud ID progetto.

• REGION: la regione principale del metastore. Per i bucket a regione singola, questa corrisponde alla regione del bucket. Per i bucket a due regioni o multiregionali, specifica una regione costituente per la replica principale. L'altra regione diventa la replica secondaria.

• DESCRIPTION: una descrizione del catalogo.

Configura Spark per utilizzare le tabelle Delta

Per utilizzare il catalogo di runtime Lakehouse con le tabelle Delta, configura la sessione Spark con la factory del client metastore Lakehouse e le estensioni di Delta Lake.

spark-sql

1. Utilizza SSH per connetterti al nodo primario del cluster Managed Service for Apache Spark.

2. Esegui spark-sql nella riga di comando con le seguenti proprietà per avviare una sessione Spark SQL interattiva:

   spark-sql \
       --conf spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension \
       --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog \
       --conf spark.hive.metastore.client.factory.class=com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory \
       --conf spark.hive.metastore.blms.project.id=PROJECT_ID \
       --conf spark.hive.metastore.blms.catalog.default=LAKEHOUSE_CATALOG_ID \
       --conf spark.hive.metastore.warehouse.dir=gs://GCS_WAREHOUSE_PATH
   

3. Dopo l'avvio della sessione, Spark si connette al catalogo di runtime Lakehouse.

   Esegui questi comandi per creare ed eseguire query sulle risorse Delta:

   -- Show all the databases in the current project.
   SHOW DATABASES;
   
   -- Create a database.
   CREATE DATABASE spark_delta_blms_database;
   
   -- Create a Delta data source table.
   CREATE TABLE spark_delta_blms_database.delta_quick_start (id INT, name STRING) USING DELTA;
   
   -- Insert data into the table.
   INSERT INTO TABLE spark_delta_blms_database.delta_quick_start VALUES (1, 'my-first-user');
   
   -- Query the Delta table.
   SELECT * FROM spark_delta_blms_database.delta_quick_start;
   

   Sostituisci quanto segue:

   • PROJECT_ID: il tuo Google Cloud progetto ID.

   • LAKEHOUSE_CATALOG_ID: il nome del catalogo Hive.

   • GCS_WAREHOUSE_PATH: il percorso Cloud Storage in cui sono archiviati i dati del data warehouse.

Blocco note Jupyter

1. Esegui un blocco note Jupyter su un cluster Managed Service for Apache Spark.
2. Accedi all'interfaccia web di Jupyter dalla scheda Interfacce web della pagina dei dettagli del cluster nella Google Cloud console.

3. In un nuovo blocco note, crea una sessione Spark configurata per Delta Lake ed esegui le query:

   from pyspark.sql import SparkSession
   
   # If a Spark session exists, stop it first by running spark.stop()
   spark = SparkSession.builder\
       .master("local")\
       .appName("Lakehouse runtime catalog Delta tutorial")\
       .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")\
       .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")\
       .config("spark.hive.metastore.client.factory.class", "com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory")\
       .config("spark.hive.metastore.blms.project.id", "PROJECT_ID")\
       .config("spark.hive.metastore.warehouse.dir", "gs://GCS_WAREHOUSE_PATH")\
       .config("spark.hive.metastore.blms.catalog.default", "LAKEHOUSE_CATALOG_ID")\
       .getOrCreate()
   
   # Show all the databases.
   df = spark.sql("SHOW DATABASES;")
   df.show()
   
   # Create a database.
   spark.sql("CREATE DATABASE jupyter_delta_blms_db")
   
   # Create a Delta data source table.
   spark.sql("CREATE TABLE jupyter_delta_blms_db.delta_table(id INT, name STRING) USING DELTA")
   
   # Insert data into the table.
   spark.sql("INSERT INTO TABLE jupyter_delta_blms_db.delta_table VALUES (1, 'my-first-user');")
   
   # Query from table.
   spark.sql("SELECT * FROM jupyter_delta_blms_db.delta_table;").show()
   

   Sostituisci quanto segue:

   • PROJECT_ID: il tuo Google Cloud progetto ID.

   • GCS_WAREHOUSE_PATH: il percorso Cloud Storage in cui sono archiviati i dati del data warehouse.

   • LAKEHOUSE_CATALOG_ID: il nome del catalogo Hive.

Blocco note BigQuery

1. Nella Google Cloud console, vai a BigQuery.

   Vai a BigQuery

2. Nel riquadro Explorer , fai clic su + AGGIUNGI e poi su Blocco note Python.

3. In una cella di codice, configura la sessione Managed Service for Apache Spark con le estensioni Delta e le proprietà del catalogo di runtime Lakehouse:

   from google.cloud.dataproc_spark_connect import DataprocSparkSession
   from google.cloud.dataproc_v1 import Session
   import os
   
   os.environ['DATAPROC_SPARK_CONNECT_DEFAULT_DATASOURCE'] = ""
   
   session = Session()
   session.environment_config.execution_config.ttl = {"seconds": 864000}
   session.runtime_config.version = "2.3"
   session.runtime_config.properties = {
    "spark.sql.extensions": "io.delta.sql.DeltaSparkSessionExtension",
    "spark.sql.catalog.spark_catalog": "org.apache.spark.sql.delta.catalog.DeltaCatalog",
    "spark.hive.metastore.blms.project.id": "PROJECT_ID",
    "spark.hive.metastore.blms.catalog.default": "LAKEHOUSE_CATALOG_ID",
    "spark.hive.metastore.warehouse.dir": "gs://GCS_WAREHOUSE_PATH",
    "spark.hive.metastore.client.factory.class": "com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory",
    "spark.sql.catalogImplementation": "hive"
   }
   
   spark = DataprocSparkSession.builder.dataprocSessionConfig(session).getOrCreate()
   print("Spark session created successfully")
   

4. In un'altra cella di codice, crea un database e una tabella Delta:

   # Create a database
   spark.sql("CREATE DATABASE bq_spark_delta_database;")
   
   # Create a Delta data source table
   spark.sql("CREATE TABLE bq_spark_delta_database.delta_quick_start (id INT, name STRING) USING DELTA;")
   
   # Insert data into the table
   spark.sql("INSERT INTO TABLE bq_spark_delta_database.delta_quick_start VALUES (1, 'my-first-user');")
   
   # Query from table
   spark.sql("SELECT * FROM bq_spark_delta_database.delta_quick_start;").show()
   

   Sostituisci quanto segue:

   • PROJECT_ID: il tuo Google Cloud progetto ID.

   • GCS_WAREHOUSE_PATH: il percorso Cloud Storage in cui sono archiviati i dati del data warehouse.

   • LAKEHOUSE_CATALOG_ID: il nome del catalogo Hive.

Invia un job batch Managed Service for Apache Spark

Puoi inviare un workload batch PySpark a Managed Service for Apache Spark per creare ed eseguire query sulle tabelle Delta con il catalogo di runtime Lakehouse.

1. Crea un file Python con un job PySpark:

   from pyspark.sql import SparkSession
   
   spark = (
       SparkSession.builder.appName("Lakehouse runtime catalog Delta tutorial")
       .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
       .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
       .config(
           "spark.hive.metastore.client.factory.class",
           "com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory",
       )
       .config("spark.hive.metastore.blms.project.id", "PROJECT_ID")
       .config("spark.hive.metastore.blms.catalog.default", "LAKEHOUSE_CATALOG_ID")
       .config(
           "spark.hive.metastore.warehouse.dir",
           "gs://GCS_WAREHOUSE_PATH",
       )
       .enableHiveSupport()
       .getOrCreate()
   )
   
   # Show all the databases.
   spark.sql("SHOW DATABASES;").show()
   
   # Create a database.
   spark.sql("CREATE DATABASE dp_serverless_delta_db")
   
   # Create a Delta data source table.
   spark.sql(
       "CREATE TABLE dp_serverless_delta_db.delta_table(id INT, name STRING) USING DELTA"
   )
   
   # Insert data into the table.
   spark.sql("INSERT INTO TABLE dp_serverless_delta_db.delta_table VALUES (1, 'my-first-user');")
   
   # Query from table.
   spark.sql("SELECT * FROM dp_serverless_delta_db.delta_table;").show()
   

   Sostituisci quanto segue:

   • PROJECT_ID: il tuo Google Cloud ID progetto.

   • LAKEHOUSE_CATALOG_ID: il nome del catalogo Hive.

   • GCS_WAREHOUSE_PATH: il percorso Cloud Storage in cui sono archiviati i dati del data warehouse.

2. Invia il job batch:

   gcloud dataproc batches submit pyspark PYTHON_SCRIPT_FILE \
       --version=2.2 \
       --project=PROJECT_ID \
       --region=REGION \
       --deps-bucket=gs://CLOUD_STORAGE_BUCKET
   

   Sostituisci quanto segue:

   • PYTHON_SCRIPT_FILE: il percorso locale o Cloud Storage del file dell'applicazione PySpark.
   • PROJECT_ID: il tuo ID progetto.
   • REGION: la regione in cui eseguire il job batch.
   • CLOUD_STORAGE_BUCKET: il bucket Cloud Storage per l'organizzazione temporanea delle dipendenze del workload.

Esegui query sulle tabelle Delta da BigQuery

Dopo aver creato le tabelle Delta da Spark nel catalogo di runtime Lakehouse, puoi eseguire query su di esse da BigQuery Studio.

Le query eseguite da BigQuery sulle tabelle Delta nell'endpoint del catalogo Hive sono di sola lettura.

1. Nella Google Cloud console, vai a BigQuery.

   Vai a BigQuery

2. Nell'editor di query, inserisci la seguente istruzione:

   SELECT * FROM `PROJECT_ID.LAKEHOUSE_CATALOG_ID.DATABASE_NAME.TABLE_NAME`;
   

   Sostituisci quanto segue:

   • PROJECT_ID: il tuo ID progetto.
   • LAKEHOUSE_CATALOG_ID: il nome del catalogo Hive.
   • DATABASE_NAME: il nome del database.
   • TABLE_NAME: il nome della tabella Delta.

Limitazioni

Quando utilizzi le tabelle Delta con l'endpoint del catalogo Hive del catalogo di runtime Lakehouse, si applicano le seguenti limitazioni:

• Query di sola lettura in BigQuery: le query sulle tabelle Delta da BigQuery sono di sola lettura. Esegui modifiche ai dati, operazioni DML e modifiche allo schema utilizzando Apache Spark.
• Schema segnaposto su GetHiveTable e ispezione dello schema reale: GetHiveTable chiamate per le tabelle Delta restituiscono uno schema segnaposto anziché lo schema della tabella effettivo. Per ispezionare lo schema reale di una tabella Delta, utilizza Apache Spark. BigQuery calcola uno schema modificato adattato per l'esecuzione di query Dremel, che non corrisponde 1:1 allo schema della tabella Delta reale.
• Incompatibilità del client Hive: i client Hive (come la CLI Hive) non supportano le tabelle Delta con l'endpoint Hive del catalogo di runtime Lakehouse. Utilizza Apache Spark per gestire le tabelle Delta e Spark o BigQuery per eseguire query su di esse.
• Operazioni gestite dal catalogo non supportate: la modalità gestita dal catalogo non è supportata per le tabelle Delta. Non puoi creare una tabella Delta utilizzando DDL metastore autonomo. Crea invece tabelle Delta con località di archiviazione esterne in Cloud Storage (ad esempio utilizzando Apache Spark) e registrale nel catalogo.
• Limitazioni delle tabelle Delta esterne di BigQuery: L' endpoint Hive del catalogo di runtime Lakehouse utilizza la stessa implementazione del motore di query sottostante delle tabelle Delta Lake esterne di BigQuery. Si applicano tutte le limitazioni delle tabelle Delta Lake esterne e le limitazioni delle tabelle BigLake. Ad esempio, le operazioni Change Data Capture (CDC) e l'API BigQuery Storage Read non sono supportate e timestamp_ntz non è supportato.

Passaggi successivi

• Informazioni sull'endpoint del catalogo Hive.
• Formati di archiviazione e tipi di dati supportati.
• Limitazioni e considerazioni per i cataloghi Hive.