Utilizzare il catalogo del runtime Lakehouse con le stored procedure Apache Spark

Le stored procedure Apache Spark eseguono i job all'interno di Lakehouse senza confini. Connettendosi al catalogo di runtime Lakehouse, queste procedure eseguono i comandi Spark SQL direttamente sul data warehouse sottostante.

Prima di iniziare

  1. Abilita la fatturazione per il tuo progetto Google Cloud . Scopri come verificare se la fatturazione è abilitata per un progetto.

  2. Abilita l'API BigQuery e l'API Dataproc.

    Abilita le API

Ruoli obbligatori

Per utilizzare le stored procedure Apache Spark, esamina i ruoli richiesti per le stored procedure e concedi i ruoli necessari.

Per ottenere le autorizzazioni necessarie per utilizzare Apache Spark e le stored procedure con il catalogo di runtime Lakehouse come servizio di metadati, chiedi all'amministratore di concederti i seguenti ruoli IAM:

  • Crea tabelle del catalogo di runtime di Lakehouse in Apache Spark:
    • BigQuery Data Editor (roles/bigquery.dataEditor) sull'account di servizio di connessione Spark nel progetto
    • Storage Object User (roles/storage.objectUser) sul account di servizio di connessione Spark nel progetto
  • Esegui query sulle tabelle del catalogo di runtime Lakehouse in BigQuery:

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.

Crea ed esegui una stored procedure

L'esempio seguente mostra come creare ed eseguire una stored procedure con il catalogo di runtime di Lakehouse.

  1. Vai alla pagina BigQuery.

    Vai a BigQuery

  2. Nell'editor di query, aggiungi il seguente codice campione per l'istruzione CREATE PROCEDURE.

    CREATE OR REPLACE PROCEDURE
`PROJECT_ID.BQ_DATASET_ID.PROCEDURE_NAME`()
WITH CONNECTION `PROJECT_ID.REGION.SPARK_CONNECTION_ID` OPTIONS (engine='SPARK',
runtime_version='1.1',
properties=[("spark.sql.catalog.CATALOG_NAME.warehouse",
"WAREHOUSE_DIRECTORY"),
("spark.sql.catalog.CATALOG_NAME.gcp_location",
"LOCATION"),
("spark.sql.catalog.CATALOG_NAME.gcp_project",
"PROJECT_ID"),
("spark.sql.catalog.CATALOG_NAME",
"org.apache.iceberg.spark.SparkCatalog"),
("spark.sql.catalog.CATALOG_NAME.catalog-impl",
"org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog"),
("spark.jars.packages",
"org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.1")],
jar_uris=["gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.6.1-1.0.2.jar"])
LANGUAGE python AS R"""
from pyspark.sql import SparkSession
spark = SparkSession \
.builder \
.appName("Lakehouse runtime catalog Iceberg") \
.getOrCreate()
spark.sql("USE CATALOG_NAME;")
spark.sql("CREATE NAMESPACE IF NOT EXISTS NAMESPACE_NAME;")
spark.sql("USE NAMESPACE_NAME;")
spark.sql("CREATE TABLE TABLE_NAME (id int, data string) USING ICEBERG LOCATION 'WAREHOUSE_DIRECTORY'")
spark.sql("DESCRIBE TABLE_NAME;")
spark.sql("INSERT INTO TABLE_NAME VALUES (1, \"first row\");")
spark.sql("SELECT * from TABLE_NAME;")
spark.sql("ALTER TABLE TABLE_NAME ADD COLUMNS (newDoubleCol double);")
spark.sql("DESCRIBE TABLE_NAME;")
""";
CALL `PROJECT_ID.BQ_DATASET_ID.PROCEDURE_NAME`();

    Sostituisci quanto segue:

    • PROJECT_ID: l'ID del tuo progetto Google Cloud .
    • BQ_DATASET_ID: l'ID del set di dati in BigQuery che contiene la procedura.
    • PROCEDURE_NAME: il nome della procedura che stai creando o sostituendo.
    • REGION: la località della tua connessione Apache Spark.
    • LOCATION: la località delle tue risorse BigQuery.
    • SPARK_CONNECTION_ID: l'ID della tua connessione Apache Spark.
    • CATALOG_NAME: il nome del catalogo che stai utilizzando.
    • WAREHOUSE_DIRECTORY: l'URI della cartella Cloud Storage che contiene il tuo data warehouse.
    • NAMESPACE_NAME: lo spazio dei nomi che stai utilizzando.

Passaggi successivi