Utilizzare BigLake Metastore con le stored procedure Spark

Questo documento spiega come utilizzare le stored procedure Apache Spark con BigLake Metastore.

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 le API BigQuery e Dataflow.

    Abilita le API

  3. (Facoltativo) Scopri di più su:

Ruoli obbligatori

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

Per ottenere le autorizzazioni necessarie per utilizzare Spark e le stored procedure con BigLake Metastore come archivio metadati, chiedi all'amministratore di concederti i seguenti ruoli IAM:

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.

Creare ed eseguire una stored procedure

Il seguente esempio mostra come creare ed eseguire una stored procedure con BigLake Metastore.

  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.1-beta.jar"])
LANGUAGE python AS R"""
from pyspark.sql import SparkSession
spark = SparkSession \
.builder \
.appName("BigLake Metastore 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 Google Cloud progetto.
    • 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 posizione della connessione Spark.
    • LOCATION: la posizione delle tue risorse BigQuery.
    • SPARK_CONNECTION_ID: l'ID della tua connessione Spark.
    • CATALOG_NAME: il nome del catalogo che stai utilizzando.
    • WAREHOUSE_DIRECTORY: l'URI della cartella Cloud Storage che contiene il data warehouse.
    • NAMESPACE_NAME: lo spazio dei nomi che stai utilizzando.

Passaggi successivi