BigQuery-Metastore mit Spark-Speicherprozeduren verwenden

In diesem Dokument wird beschrieben, wie Sie Apache Spark-Speicherprozeduren mit dem BigQuery-Metaspeicher verwenden.

Hinweise

  1. Aktivieren Sie die Abrechnung für Ihr Google Cloud -Projekt. So prüfen Sie, ob die Abrechnung für ein Projekt aktiviert ist

  2. Aktivieren Sie die BigQuery- und Dataflow APIs.

    APIs aktivieren

  3. Optional: Weitere Informationen zu folgenden Themen:

Erforderliche Rollen

Wenn Sie Spark-gespeicherte Prozeduren verwenden möchten, sehen Sie sich die erforderlichen Rollen für gespeicherte Prozeduren an und gewähren Sie die erforderlichen Rollen.

Bitten Sie Ihren Administrator, Ihnen die folgenden IAM-Rollen zuzuweisen, um die Berechtigungen zu erhalten, die Sie zum Verwenden von Spark und gespeicherten Prozeduren mit dem BigQuery-Metadatenspeicher als Metadatenspeicher benötigen:

Weitere Informationen zum Zuweisen von Rollen finden Sie unter Zugriff auf Projekte, Ordner und Organisationen verwalten.

Sie können die erforderlichen Berechtigungen auch über benutzerdefinierte Rollen oder andere vordefinierte Rollen erhalten.

Gespeicherte Prozedur erstellen und ausführen

Im folgenden Beispiel wird gezeigt, wie Sie eine gespeicherte Prozedur mit dem BigQuery-Metastore erstellen und ausführen.

  1. Rufen Sie die Seite BigQuery auf.

    BigQuery aufrufen

  2. Fügen Sie im Abfrageeditor den folgenden Beispielcode für die CREATE PROCEDURE-Anweisung hinzu.

    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`();

    Ersetzen Sie Folgendes:

    • PROJECT_ID: die ID Ihres Google Cloud -Projekts.
    • BQ_DATASET_ID: die ID des Datasets in BigQuery, das die Prozedur enthält.
    • PROCEDURE_NAME: Der Name des Verfahrens, das Sie erstellen oder ersetzen.
    • REGION: den Speicherort Ihrer Spark-Verbindung.
    • LOCATION: den Speicherort Ihrer BigQuery-Ressourcen.
    • SPARK_CONNECTION_ID: die ID Ihrer Spark-Verbindung.
    • CATALOG_NAME: der Name des verwendeten Katalogs.
    • WAREHOUSE_DIRECTORY: der URI des Cloud Storage-Ordners, der Ihr Data Warehouse enthält.
    • NAMESPACE_NAME: der von Ihnen verwendete Namespace.

Nächste Schritte