Questa pagina è stata tradotta dall'API Cloud Translation.

Utilizzare il metastore BigQuery con le stored procedure Spark

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

Prima di iniziare

Attiva la fatturazione per il tuo progetto Google Cloud . Scopri come controllare se la fatturazione è attivata in un progetto.
Abilita le API BigQuery e Dataflow.

Abilita le API
(Facoltativo) Scopri di più su:
- Scopri come funziona il metastore BigQuery e perché dovresti usarlo.
- Scopri come funzionano le stored procedure BigQuery Spark e completa le attività preliminari.

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 BigQuery come archivio dei metadati, chiedi all'amministratore di concederti i seguenti ruoli IAM:

Crea le tabelle del metastore BigQuery in Spark:
- Editor dati BigQuery (roles/bigquery.dataEditor) nell'account di servizio della connessione Spark nel progetto
- Amministratore oggetti Storage (roles/storage.objectAdmin) nell'account di servizio Spark Connection del progetto
Esegui query sulle tabelle del metastore BigQuery in BigQuery:
- Visualizzatore dati BigQuery (roles/bigquery.dataViewer) nel progetto
- Utente BigQuery (roles/bigquery.user) nel progetto
- Visualizzatore oggetti Storage (roles/storage.objectViewer) nel progetto

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

Il seguente esempio mostra come creare ed eseguire una stored procedure con il metastore BigQuery.

Vai alla pagina BigQuery.

Vai a BigQuery

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.3_2.12:1.5.2")],
jar_uris=["gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.5.2-1.0.0-beta.jar"])
LANGUAGE python AS R"""
from pyspark.sql import SparkSession
spark = SparkSession \
.builder \
.appName("BigQuery 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 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 posizione della connessione Spark.
LOCATION: la posizione delle 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.