BigQuery-Metastore mit Tabellen in BigQuery verwenden

In diesem Dokument wird erläutert, 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 Iceberg und externe BigLake-Tabellen für Apache Iceberg in 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-Metaspeichers und dazu, warum Sie ihn verwenden sollten.

Erforderliche Rollen

Bitten Sie Ihren Administrator, Ihnen die folgenden IAM-Rollen zuzuweisen, um Spark und Dataproc mit dem BigQuery-Metastore als Metadatenspeicher zu verwenden:

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 in der Google Cloud Console ein Dataset.

    CREATE SCHEMA `PROJECT_ID`.DATASET_NAME;

    Ersetzen Sie Folgendes:

    • PROJECT_ID: die ID des Google Cloud Projekts, in dem das Dataset erstellt werden soll.
    • DATASET_NAME: Ein Name für Ihr Dataset.

  2. Erstellen Sie eine Cloud-Ressourcenverbindung.

  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 Ihre Tabelle.

  4. Fügen Sie Daten in die Standard-BigQuery-Tabelle ein.

    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 Ihre BigQuery-Tabelle für Apache Iceberg. 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. Fügen Sie Daten in die BigQuery-Tabelle für Apache Iceberg ein.

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

  7. Erstellen Sie eine BigLake-externe Tabelle für Apache Iceberg.

    Wenn Sie beispielsweise eine BigLake-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: Ein Name für die schreibgeschützte 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 BigQuery-Tabelle für Apache Iceberg und die BigLake-externe Tabelle für Apache Iceberg aus.

    from pyspark.sql import SparkSession

# Create a spark session
spark = SparkSession.builder \
.appName("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 the tables
sql = """SELECT * FROM DATASET_NAME.TABLE_NAME"""
df = spark.read.format("bigquery").load(sql)
df.show()

sql = """SELECT * FROM DATASET_NAME.ICEBERG_TABLE_NAME"""
df = spark.read.format("bigquery").load(sql)
df.show()

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 mit Ihrer Iceberg-Tabelle und Ihrer BigLake Iceberg-Tabelle verknüpft ist.
    • CATALOG_NAME: der Name des verwendeten Katalogs.
    • MATERIALIZATION_NAMESPACE: der Namespace für das Speichern temporärer Ergebnisse.

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

    gcloud dataproc batches submit pyspark SCRIPT_PATH \
  --version=2.2 \
  --project=PROJECT_ID \
  --region=REGION \
  --deps-bucket=YOUR_BUCKET \

    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-Buckets, in den Arbeitslastabhängigkeiten 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