BigQuery-Metastore mit Tabellen in BigQuery verwenden

In diesem Dokument wird beschrieben, wie Sie den BigQuery-Metastore mit BigQuery-Tabellen und Spark verwenden.

Mit dem BigQuery-Metastore können Sie standardmäßige (integrierte) Tabellen, BigQuery-Tabellen für Apache Spark Iceberg und BigLake-externe Tabellen aus BigQuery erstellen und 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 Dataproc APIs.

    APIs aktivieren

  3. Optional: Informationen zur Funktionsweise des BigQuery-Metastores und dazu, warum Sie ihn verwenden sollten.

Erforderliche Rollen

Bitten Sie Ihren Administrator, Ihnen die folgenden IAM-Rollen zuzuweisen, um die Berechtigungen zu erhalten, die Sie zum Verwenden von Spark und Dataproc mit BigQuery Metastore 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.

Verbindung zu einer Tabelle herstellen

  1. Erstellen Sie einen Datensatz in der Google Cloud Console.

    CREATE SCHEMA `PROJECT_ID`.DATASET_NAME;

    Ersetzen Sie Folgendes:

    • PROJECT_ID: Die ID des Google Cloud -Projekts, in dem der Datensatz erstellt werden soll.
    • DATASET_NAME: ein Name für das Dataset.

  2. Erstellen Sie eine Cloud-Ressourcen-Verbindung.

  3. Erstellen Sie eine BigQuery-Standardtabelle.

    CREATE TABLE `PROJECT_ID`.DATASET_NAME.TABLE_NAME (name STRING,id INT64);

    Ersetzen Sie Folgendes:

    • TABLE_NAME: ein Name für die Tabelle.

  4. Daten in die BigQuery-Standardtabelle einfügen

    INSERT INTO `PROJECT_ID`.DATASET_NAME.TABLE_NAME VALUES ('test_name1', 123),('test_name2', 456),('test_name3', 789);

  5. Erstellen Sie eine BigQuery-Tabelle für Apache Iceberg.

    Wenn Sie beispielsweise eine Tabelle erstellen möchten, führen Sie die folgende CREATE-Anweisung aus.

    CREATE TABLE `PROJECT_ID`.DATASET_NAME.ICEBERG_TABLE_NAME(
name STRING,id INT64
)
WITH CONNECTION `CONNECTION_NAME`
OPTIONS (
file_format = 'PARQUET',
table_format = 'ICEBERG',
storage_uri = 'STORAGE_URI');

    Ersetzen Sie Folgendes:

    • ICEBERG_TABLE_NAME: ein Name für die Iceberg-Tabelle. Beispiel: iceberg_managed_table.
    • CONNECTION_NAME: Der Name der Verbindung. Sie haben diese im vorherigen Schritt erstellt. Beispiel: myproject.us.myconnection.
    • STORAGE_URI: ein vollständig qualifizierter Cloud Storage-URI. Beispiel: gs://mybucket/table.

  6. Daten in die BigQuery-Tabelle für Apache Iceberg einfügen

    INSERT INTO `PROJECT_ID`.DATASET_NAME.ICEBERG_TABLE_NAME VALUES ('test_name1', 123),('test_name2', 456),('test_name3', 789);

  7. Erstellen Sie eine schreibgeschützte Iceberg-Tabelle.

    Wenn Sie beispielsweise eine schreibgeschützte Iceberg-Tabelle erstellen möchten, führen Sie die folgende CREATE-Anweisung aus.

    CREATE OR REPLACE EXTERNAL TABLE  `PROJECT_ID`.DATASET_NAME.READONLY_ICEBERG_TABLE_NAME
WITH CONNECTION `CONNECTION_NAME`
OPTIONS (
  format = 'ICEBERG',
  uris =
    ['BUCKET_PATH'],
  require_partition_filter = FALSE);

    Ersetzen Sie Folgendes:

    • READONLY_ICEBERG_TABLE_NAME: Name der schreibgeschützten Tabelle.
    • BUCKET_PATH: der Pfad zum Cloud Storage-Bucket, der die Daten für die externe Tabelle im Format ['gs://bucket_name/[folder_name/]file_name'] enthält.

  8. Führen Sie in PySpark Abfragen für die Standardtabelle, die verwaltete Iceberg-Tabelle und die schreibgeschützte Iceberg-Tabelle aus.

    from pyspark.sql import SparkSession

# Create a spark session
spark = SparkSession.builder \
.appName("BigQuery Metastore Iceberg") \
.config("spark.sql.catalog.CATALOG_NAME", "org.apache.iceberg.spark.SparkCatalog") \
.config("spark.sql.catalog.CATALOG_NAME.catalog-impl", "org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog") \
.config("spark.sql.catalog.CATALOG_NAME.gcp_project", "PROJECT_ID") \
.config("spark.sql.catalog.CATALOG_NAME.gcp_location", "LOCATION") \
.config("spark.sql.catalog.CATALOG_NAME.warehouse", "WAREHOUSE_DIRECTORY") \
.getOrCreate()
spark.conf.set("viewsEnabled","true")

# Use the bqms_catalog
spark.sql("USE `CATALOG_NAME`;")
spark.sql("USE NAMESPACE DATASET_NAME;")

# Configure spark for temp results
spark.sql("CREATE namespace if not exists MATERIALIZATION_NAMESPACE");
spark.conf.set("materializationDataset","MATERIALIZATION_NAMESPACE")

# List the tables in the dataset
df = spark.sql("SHOW TABLES;")
df.show();

# Query a standard BigQuery table
sql = """SELECT * FROM DATASET_NAME.TABLE_NAME"""
df = spark.read.format("bigquery").load(sql)
df.show()

# Query a BigQuery Managed Apache Iceberg table
sql = """SELECT * FROM DATASET_NAME.ICEBERG_TABLE_NAME"""
df = spark.read.format("bigquery").load(sql)
df.show()

# Query a BigQuery Readonly Apache Iceberg table
sql = """SELECT * FROM DATASET_NAME.READONLY_ICEBERG_TABLE_NAME"""
df = spark.read.format("bigquery").load(sql)
df.show()

    Ersetzen Sie Folgendes:

    • WAREHOUSE_DIRECTORY: der URI des Cloud Storage-Ordners, der Ihr Data Warehouse enthält.
    • CATALOG_NAME: der Name des verwendeten Katalogs.
    • MATERIALIZATION_NAMESPACE: der Namespace zum Speichern temporärer Ergebnisse.

  9. Führen Sie das PySpark-Script mit Serverless Spark aus.

    gcloud dataproc batches submit pyspark SCRIPT_PATH \
  --version=2.2 \
  --project=PROJECT_ID \
  --region=REGION \
  --deps-bucket=YOUR_BUCKET \
  --jars=https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.5.2/iceberg-spark-runtime-3.5_2.12-1.5.2.jar,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.5.2-1.0.0-beta.jar

    Ersetzen Sie Folgendes:

    • SCRIPT_PATH: Der Pfad zum Script, das vom Batchjob verwendet wird.
    • PROJECT_ID: Die ID des Google Cloud -Projekts, in dem der Batchjob ausgeführt werden soll.
    • REGION: die Region, in der Ihre Arbeitslast ausgeführt wird.
    • YOUR_BUCKET: der Speicherort des Cloud Storage-Bucket, in den die Abhängigkeiten der Arbeitslast hochgeladen werden sollen. Das gs://-URI-Präfix des Buckets ist nicht erforderlich. Sie können den Bucket-Pfad oder den Bucket-Namen angeben, z. B. mybucketname1.

Nächste Schritte