Spark mit Delta-Tabellen einrichten

In diesem Dokument wird erläutert, wie Sie Apache Spark mit dem Lakehouse-Laufzeitkatalog konfigurieren, um Delta Lake-Tabellen zu verwenden. Sie erfahren, wie Sie einen Apache Hive-Katalog erstellen, Spark-Sitzungen so konfigurieren, dass sie eine Verbindung zum Metastore herstellen, und Arbeitslasten ausführen, um Delta-Tabellen zu erstellen und zu registrieren, die in BigQuery abgefragt werden können.

Hinweis: Sie können keine neue Delta-Tabelle direkt über den Hive-Endpunkt des Lakehouse-Laufzeitkatalogs oder mit Hive DDL erstellen. Verwenden Sie stattdessen eine Delta-kompatible Engine wie Apache Spark, um die zugrunde liegenden Tabellendateien in Cloud Storage zu erstellen und die Tabelle in der Metastore zu registrieren.

Hinweis

1. Informationen zum Hive-Katalogendpunkt
2. Überprüfen Sie Unterstützte Speicherformate und Datentypen.
3. Überprüfen Sie Einschränkungen und Überlegungen.
Melden Sie sich in Ihrem Google Cloud Konto an. Wenn Sie noch kein Konto haben Google Cloud, erstellen Sie ein Konto, um die Leistungsfähigkeit unserer Produkte in der Praxis zu sehen und zu bewerten. Neukunden erhalten außerdem ein Guthaben von 300 $, um Arbeitslasten auszuführen, zu testen und bereitzustellen.

Verify that billing is enabled for your Google Cloud project.

Enable the Lakehouse, Managed Service for Apache Spark APIs, if any are not already enabled.

Roles required to enable APIs

To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

Enable the APIs

Verify that billing is enabled for your Google Cloud project.

Enable the Lakehouse, Managed Service for Apache Spark APIs, if any are not already enabled.

Roles required to enable APIs

To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

Enable the APIs

Erforderliche Rollen

Bitten Sie Ihren Administrator, Ihnen die folgenden IAM-Rollen für Ihr Projekt zuzuweisen, um die Berechtigungen zu erhalten, die Sie zur Verwendung des Lakehouse-Laufzeitkatalogs mit Delta-Tabellen benötigen:

• Managed Service for Apache Spark-Cluster erstellen: Dataproc-Bearbeiter (roles/dataproc.editor)
• Dienstkonto des Clusters:
  • Dataproc-Worker (roles/dataproc.worker)
  • Storage Object User (roles/storage.objectUser)
  • BigLake-Bearbeiter (roles/biglake.editor)
  • Service Usage Consumer (roles/serviceusage.serviceUsageConsumer)
• Schreibzugriff auf alle Ressourcen des Lakehouse-Laufzeitkatalogs: BigLake-Bearbeiter (roles/biglake.editor)
• Schreibgeschützter Zugriff auf alle Ressourcen des Lakehouse-Laufzeitkatalogs: BigLake-Betrachter (roles/biglake.viewer)

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.

Eine Anleitung finden Sie unter Einzelne Rolle zuweisen.

Allgemeiner Workflow

So verwenden Sie den Lakehouse-Laufzeitkatalog mit Delta-Tabellen:

1. Erstellen Sie einen Lakehouse Hive-Katalog.
2. Konfigurieren Sie Ihre Spark-Sitzung mit dem Lakehouse-Metastore-Client und den Delta Lake-Abhängigkeiten.
3. Erstellen und füllen Sie Delta-Tabellen in Cloud Storage mit Spark. Dadurch werden sie im Metastore registriert.
4. Fragen Sie die Delta-Tabellen über Apache Spark (Lesen und Schreiben) oder direkt über BigQuery (schreibgeschützt) ab.

Lakehouse Hive-Katalog erstellen

Wenn Sie den Lakehouse-Laufzeitkatalog mit Spark und Delta-Tabellen verwenden möchten, erstellen Sie einen Hive-Katalog.

Ein Lakehouse Hive-Katalog ist eine Sammlung von Hive-Datenbanken. Erstellen Sie vor dem Ausführen von Spark-Jobs einen Katalog, um ihn im Lakehouse-Metastore zu registrieren. Für den Katalog sind ein Name und ein Cloud Storage-Speicherort für die Daten erforderlich.

Console

1. Öffnen Sie in der Google Cloud Console die Seite Lakehouse.

   Lakehouse aufrufen

2. Klicken Sie auf Katalog erstellen.

3. Wählen Sie Lakehouse-Laufzeitkatalog aus.

4. Wählen Sie unter Katalogtyp die Option Hive Metastore aus.

5. Geben Sie im Feld Cloud Storage-Bucket auswählen den Namen des Cloud Storage-Buckets für Ihren Katalog ein. Alternativ können Sie auf Durchsuchen klicken, um einen Bucket auszuwählen oder zu erstellen.

6. Geben Sie unter Katalog-ID einen Namen für Ihren Lakehouse Hive Katalog ein.

7. Geben Sie unter Primärer Speicherort dieselbe Region wie für Ihren Bucket an.

8. Klicken Sie auf Erstellen.

gcloud

Führen Sie den folgenden Befehl aus, um einen Hive-Katalog zu erstellen:

gcloud biglake hive catalogs create LAKEHOUSE_CATALOG_ID \
    --project=PROJECT_ID \
    --location-uri="gs://GCS_WAREHOUSE_PATH" \
    --primary-location=REGION \
    --description="DESCRIPTION"

Dabei gilt:

• LAKEHOUSE_CATALOG_ID: der Name des Hive-Katalogs.

• GCS_WAREHOUSE_PATH: der Cloud Storage-Pfad für Ihre Warehouse-Daten.

• PROJECT_ID: Ihre Google Cloud Projekt-ID.

• REGION: die primäre Region des Metastores. Bei Buckets mit einer einzelnen Region entspricht dies der Bucket-Region. Geben Sie für Buckets mit zwei oder mehreren Regionen eine Region für das primäre Replikat an. Die andere Region wird zum sekundären Replikat.

• DESCRIPTION: eine Beschreibung des Katalogs.

Spark für die Verwendung von Delta-Tabellen konfigurieren

Wenn Sie den Lakehouse-Laufzeitkatalog mit Delta-Tabellen verwenden möchten, konfigurieren Sie Ihre Spark-Sitzung mit der Lakehouse-Metastore-Client-Factory und den Delta Lake-Erweiterungen.

spark-sql

1. Stellen Sie über SSH eine Verbindung zum primären Knoten Ihres Managed Service for Apache Spark-Clusters her.

2. Führen Sie spark-sql in der Befehlszeile mit den folgenden Eigenschaften aus, um eine interaktive Spark SQL-Sitzung zu starten:

   spark-sql \
       --conf spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension \
       --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog \
       --conf spark.hive.metastore.client.factory.class=com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory \
       --conf spark.hive.metastore.blms.project.id=PROJECT_ID \
       --conf spark.hive.metastore.blms.catalog.default=LAKEHOUSE_CATALOG_ID \
       --conf spark.hive.metastore.warehouse.dir=gs://GCS_WAREHOUSE_PATH
   

3. Nach dem Start der Sitzung stellt Spark eine Verbindung zum Lakehouse-Laufzeitkatalog her.

   Führen Sie die folgenden Befehle aus, um Delta-Ressourcen zu erstellen und abzufragen:

   -- Show all the databases in the current project.
   SHOW DATABASES;
   
   -- Create a database.
   CREATE DATABASE spark_delta_blms_database;
   
   -- Create a Delta data source table.
   CREATE TABLE spark_delta_blms_database.delta_quick_start (id INT, name STRING) USING DELTA;
   
   -- Insert data into the table.
   INSERT INTO TABLE spark_delta_blms_database.delta_quick_start VALUES (1, 'my-first-user');
   
   -- Query the Delta table.
   SELECT * FROM spark_delta_blms_database.delta_quick_start;
   

   Dabei gilt:

   • PROJECT_ID: Ihre Google Cloud Projekt ID.

   • LAKEHOUSE_CATALOG_ID: der Name des Hive-Katalogs.

   • GCS_WAREHOUSE_PATH: der Cloud Storage-Pfad, in dem Ihre Warehouse-Daten gespeichert sind.

Jupyter-Notebook

1. Führen Sie ein Jupyter-Notebook in einem Managed Service for Apache Spark-Cluster aus.
2. Greifen Sie über den Tab Weboberflächen auf der Detailseite des Clusters in der Google Cloud Console auf die Jupyter-Weboberfläche zu.

3. Erstellen Sie in einem neuen Notebook eine für Delta Lake konfigurierte Spark-Sitzung und führen Sie Abfragen aus:

   from pyspark.sql import SparkSession
   
   # If a Spark session exists, stop it first by running spark.stop()
   spark = SparkSession.builder\
       .master("local")\
       .appName("Lakehouse runtime catalog Delta tutorial")\
       .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")\
       .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")\
       .config("spark.hive.metastore.client.factory.class", "com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory")\
       .config("spark.hive.metastore.blms.project.id", "PROJECT_ID")\
       .config("spark.hive.metastore.warehouse.dir", "gs://GCS_WAREHOUSE_PATH")\
       .config("spark.hive.metastore.blms.catalog.default", "LAKEHOUSE_CATALOG_ID")\
       .getOrCreate()
   
   # Show all the databases.
   df = spark.sql("SHOW DATABASES;")
   df.show()
   
   # Create a database.
   spark.sql("CREATE DATABASE jupyter_delta_blms_db")
   
   # Create a Delta data source table.
   spark.sql("CREATE TABLE jupyter_delta_blms_db.delta_table(id INT, name STRING) USING DELTA")
   
   # Insert data into the table.
   spark.sql("INSERT INTO TABLE jupyter_delta_blms_db.delta_table VALUES (1, 'my-first-user');")
   
   # Query from table.
   spark.sql("SELECT * FROM jupyter_delta_blms_db.delta_table;").show()
   

   Dabei gilt:

   • PROJECT_ID: Ihre Google Cloud Projekt ID.

   • GCS_WAREHOUSE_PATH: der Cloud Storage-Pfad, in dem Ihre Warehouse-Daten gespeichert sind.

   • LAKEHOUSE_CATALOG_ID: der Name des Hive-Katalogs.

BigQuery-Notebook

1. Wechseln Sie in der Google Cloud Console zu BigQuery.

   BigQuery aufrufen

2. Klicken Sie im Bereich Explorer auf + HINZUFÜGEN und dann auf Python Notebook.

3. Konfigurieren Sie in einer Codezelle die Managed Service for Apache Spark-Sitzung mit Delta-Erweiterungen und Lakehouse-Laufzeitkatalogeigenschaften:

   from google.cloud.dataproc_spark_connect import DataprocSparkSession
   from google.cloud.dataproc_v1 import Session
   import os
   
   os.environ['DATAPROC_SPARK_CONNECT_DEFAULT_DATASOURCE'] = ""
   
   session = Session()
   session.environment_config.execution_config.ttl = {"seconds": 864000}
   session.runtime_config.version = "2.3"
   session.runtime_config.properties = {
    "spark.sql.extensions": "io.delta.sql.DeltaSparkSessionExtension",
    "spark.sql.catalog.spark_catalog": "org.apache.spark.sql.delta.catalog.DeltaCatalog",
    "spark.hive.metastore.blms.project.id": "PROJECT_ID",
    "spark.hive.metastore.blms.catalog.default": "LAKEHOUSE_CATALOG_ID",
    "spark.hive.metastore.warehouse.dir": "gs://GCS_WAREHOUSE_PATH",
    "spark.hive.metastore.client.factory.class": "com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory",
    "spark.sql.catalogImplementation": "hive"
   }
   
   spark = DataprocSparkSession.builder.dataprocSessionConfig(session).getOrCreate()
   print("Spark session created successfully")
   

4. Erstellen Sie in einer anderen Codezelle eine Datenbank und eine Delta-Tabelle:

   # Create a database
   spark.sql("CREATE DATABASE bq_spark_delta_database;")
   
   # Create a Delta data source table
   spark.sql("CREATE TABLE bq_spark_delta_database.delta_quick_start (id INT, name STRING) USING DELTA;")
   
   # Insert data into the table
   spark.sql("INSERT INTO TABLE bq_spark_delta_database.delta_quick_start VALUES (1, 'my-first-user');")
   
   # Query from table
   spark.sql("SELECT * FROM bq_spark_delta_database.delta_quick_start;").show()
   

   Dabei gilt:

   • PROJECT_ID: Ihre Google Cloud Projekt ID.

   • GCS_WAREHOUSE_PATH: der Cloud Storage-Pfad, in dem Ihre Warehouse-Daten gespeichert sind.

   • LAKEHOUSE_CATALOG_ID: der Name des Hive-Katalogs.

Batchjob für Managed Service for Apache Spark senden

Sie können eine PySpark-Batcharbeitslast an Managed Service for Apache Spark senden, um mit dem Lakehouse-Laufzeitkatalog Delta-Tabellen zu erstellen und abzufragen.

1. Erstellen Sie eine Python-Datei mit einem PySpark-Job:

   from pyspark.sql import SparkSession
   
   spark = (
       SparkSession.builder.appName("Lakehouse runtime catalog Delta tutorial")
       .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
       .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
       .config(
           "spark.hive.metastore.client.factory.class",
           "com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory",
       )
       .config("spark.hive.metastore.blms.project.id", "PROJECT_ID")
       .config("spark.hive.metastore.blms.catalog.default", "LAKEHOUSE_CATALOG_ID")
       .config(
           "spark.hive.metastore.warehouse.dir",
           "gs://GCS_WAREHOUSE_PATH",
       )
       .enableHiveSupport()
       .getOrCreate()
   )
   
   # Show all the databases.
   spark.sql("SHOW DATABASES;").show()
   
   # Create a database.
   spark.sql("CREATE DATABASE dp_serverless_delta_db")
   
   # Create a Delta data source table.
   spark.sql(
       "CREATE TABLE dp_serverless_delta_db.delta_table(id INT, name STRING) USING DELTA"
   )
   
   # Insert data into the table.
   spark.sql("INSERT INTO TABLE dp_serverless_delta_db.delta_table VALUES (1, 'my-first-user');")
   
   # Query from table.
   spark.sql("SELECT * FROM dp_serverless_delta_db.delta_table;").show()
   

   Dabei gilt:

   • PROJECT_ID: Ihre Google Cloud Projekt-ID.

   • LAKEHOUSE_CATALOG_ID: der Name des Hive-Katalogs.

   • GCS_WAREHOUSE_PATH: der Cloud Storage-Pfad, in dem Ihre Warehouse-Daten gespeichert sind.

2. Senden Sie den Batchjob:

   gcloud dataproc batches submit pyspark PYTHON_SCRIPT_FILE \
       --version=2.2 \
       --project=PROJECT_ID \
       --region=REGION \
       --deps-bucket=gs://CLOUD_STORAGE_BUCKET
   

   Dabei gilt:

   • PYTHON_SCRIPT_FILE: der lokale oder Cloud Storage-Pfad zur PySpark-Anwendungsdatei.
   • PROJECT_ID: Ihre Projekt-ID.
   • REGION: die Region, in der der Batchjob ausgeführt werden soll.
   • CLOUD_STORAGE_BUCKET: der Cloud Storage-Bucket für das Staging von Arbeitslastabhängigkeiten.

Delta-Tabellen aus BigQuery abfragen

Nachdem Sie Delta-Tabellen aus Spark im Lakehouse-Laufzeitkatalog erstellt haben, können Sie sie in BigQuery Studio abfragen.

Abfragen, die in BigQuery für Delta-Tabellen im Hive-Katalogendpunkt ausgeführt werden, sind schreibgeschützt.

1. Wechseln Sie in der Google Cloud Console zu BigQuery.

   BigQuery aufrufen

2. Geben Sie im Abfrageeditor die folgende Anweisung ein:

   SELECT * FROM `PROJECT_ID.LAKEHOUSE_CATALOG_ID.DATABASE_NAME.TABLE_NAME`;
   

   Dabei gilt:

   • PROJECT_ID: Ihre Projekt-ID.
   • LAKEHOUSE_CATALOG_ID: der Name Ihres Hive-Katalogs.
   • DATABASE_NAME: der Name Ihrer Datenbank.
   • TABLE_NAME: der Name Ihrer Delta-Tabelle.

Beschränkungen

Bei der Verwendung von Delta-Tabellen mit dem Hive-Katalogendpunkt des Lakehouse-Laufzeitkatalogs gelten die folgenden Einschränkungen:

• Lesezugriff-Abfragen in BigQuery: Abfragen für Delta-Tabellen aus BigQuery sind schreibgeschützt. Nehmen Sie Datenänderungen, DML-Vorgänge und Schemaänderungen mit Apache Spark vor.
• Platzhalterschema für GetHiveTable und echte Schemaüberprüfung: GetHiveTable-Aufrufe für Delta-Tabellen geben ein Platzhalterschema anstelle von dem tatsächlichen Tabellenschema zurück. Verwenden Sie Apache Spark, um das tatsächliche Schema einer Delta-Tabelle zu prüfen. BigQuery berechnet ein geändertes Schema, das für die Dremel-Abfrageausführung angepasst ist und nicht mit dem tatsächlichen Delta-Tabellenschema übereinstimmt.
• Inkompatibilität des Hive-Clients: Hive-Clients (z. B. die Hive-CLI) unterstützen keine Delta-Tabellen mit dem Hive-Endpunkt des Lakehouse-Laufzeitkatalogs. Verwenden Sie Apache Spark, um Delta-Tabellen zu verwalten, und Spark oder BigQuery, um sie abzufragen.
• Vom Katalog verwaltete Vorgänge werden nicht unterstützt: Der vom Katalog verwaltete Modus wird für Delta-Tabellen nicht unterstützt. Sie können keine Delta-Tabelle mit eigenständiger Metastore-DDL erstellen. Erstellen Sie stattdessen Delta-Tabellen mit externen Speicherorten in Cloud Storage (z. B. mit Apache Spark) und registrieren Sie sie im Katalog.
• Einschränkungen für externe Delta-Tabellen in BigQuery: Der Hive-Endpunkt des Lakehouse-Laufzeitkatalogs verwendet dieselbe zugrunde liegende Abfrage-Engine-Implementierung wie externe Delta Lake-Tabellen in BigQuery. Alle Einschränkungen für externe Delta Lake-Tabellen und BigLake-Tabellen gelten. Beispielsweise werden Change Data Capture-Vorgänge (CDC) und die BigQuery Storage Read API nicht unterstützt und timestamp_ntz wird nicht unterstützt.

Nächste Schritte

• Informationen zum Hive-Katalogendpunkt.
• Unterstützte Speicherformate und Datentypen.
• Einschränkungen und Überlegungen für Hive-Kataloge.