Utilizzare il metastore BigLake con le stored procedure Spark

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

Prima di iniziare

  1. Abilita la fatturazione per il tuo Google Cloud progetto. Scopri come verificare se la fatturazione è attivata in 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 di Spark, esamina i ruoli richiesti per le stored procedure e concedi i ruoli necessari.

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

Per saperne di più sulla concessione dei ruoli, consulta Gestire 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

L'esempio seguente mostra come creare ed eseguire una procedura memorizzata con il metastore BigLake.

  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 connessione Spark.
    • CATALOG_NAME: il nome del catalogo che stai utilizzando.
    • WAREHOUSE_DIRECTORY: l'URI della cartella Cloud Storage contenente il data warehouse.
    • NAMESPACE_NAME: lo spazio dei nomi in uso.

Passaggi successivi