Spark und Hive mit dem Lakehouse-Laufzeitkatalog einrichten

Mit dem Lakehouse-Laufzeitkatalog können Sie Apache Spark- und Apache Hive-Arbeitslasten mit einheitlichen Metadaten ausführen, sodass Sie Ihre Tabellen direkt in BigQuery abfragen können. Zum Verwalten dieser Arbeitslasten erstellen Sie einen Apache Hive-Katalogendpunkt, konfigurieren Ihre Spark-Sitzungen für die Verbindung mit dem Metastore und führen Spark-Jobs aus, um abfragbare Tabellen zu generieren.

Hinweis

  1. Weitere Informationen dazu, wie Spark eine Verbindung zum Lakehouse-Laufzeitkatalog herstellt, finden Sie unter Hive-Kataloge im Lakehouse-Laufzeitkatalog.
  2. Lesen Sie die Informationen unter Unterstützte Speicherformate und Datentypen.
  3. Lesen Sie die Einschränkungen und Überlegungen.
  4. Melden Sie sich in Ihrem Google Cloud -Konto an. Wenn Sie mit Google Cloudnoch nicht vertraut sind, erstellen Sie einfach ein Konto, um die Leistungsfähigkeit unserer Produkte in der Praxis sehen und bewerten zu können. Neukunden erhalten außerdem ein Guthaben von 300 $, um Arbeitslasten auszuführen, zu testen und bereitzustellen.

  5. Verify that billing is enabled for your Google Cloud project.

  6. Enable the BigLake, 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

  7. Verify that billing is enabled for your Google Cloud project.

  8. Enable the BigLake, 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

    9.

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

Eine Anleitung finden Sie unter Einzelne Rolle zuweisen.

Allgemeiner Workflow

Wenn Sie den Lakehouse-Laufzeitkatalog mit Spark und Hive verwenden möchten, gehen Sie so vor:

  1. Erstellen Sie einen Lakehouse-Hive-Katalogendpunkt.
  2. Konfigurieren Sie Ihre Spark-Sitzung mit dem von Ihnen bevorzugten Tool (z. B. Managed Service for Apache Spark oder BigQuery Studio).
  3. Datenbank- und Tabellenvorgänge in Ihrer Spark-Sitzung ausführen
  4. Senden Sie Batcharbeitslasten an Managed Service for Apache Spark und fragen Sie die resultierenden Tabellen direkt in BigQuery ab.

Lakehouse Hive-Katalog erstellen

Wenn Sie den Lakehouse-Laufzeitkatalog mit Spark und Hive verwenden möchten, müssen Sie zuerst einen Hive-Katalog erstellen.

Ein Lakehouse-Hive-Katalog ist eine Sammlung von Hive-Datenbanken. Bevor Sie Spark-Jobs ausführen, erstellen Sie einen Katalog, um ihn im Lakehouse-Laufzeitkatalog zu registrieren. Der Katalog hat einen Namen und einen Cloud Storage-Speicherort, an dem sich die Hive-Daten befinden.

Console

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

    Zu Lakehouse

  2. Klicken Sie auf Katalog erstellen.

  3. Wählen Sie Lakehouse-Laufzeitkatalog aus.

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

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

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

  7. Geben Sie für Primärer Standort 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"

Ersetzen Sie Folgendes:

  • LAKEHOUSE_CATALOG_ID: der Name des Hive-Katalogs.

  • GCS_WAREHOUSE_PATH: der Cloud Storage-Pfad, in dem Ihr Hive-Warehouse gespeichert ist.

  • PROJECT_ID: Ihre Google Cloud-Projekt-ID.

  • REGION: die primäre Region des Metastores. Bei Buckets mit einer einzelnen Region sollte er mit der Bucket-Region übereinstimmen. Bei biregionalen oder multiregionalen Buckets sollte es sich um eine der zugehörigen Regionen handeln, in der sich das primäre Replikat des Metaspeichers befinden soll. Die andere Region wird zum sekundären Replikat.

  • DESCRIPTION: Eine Beschreibung des Katalogs.

curl

  1. Führen Sie den folgenden Befehl aus, um einen Hive-Katalog zu erstellen:
curl -X POST -s -i -H "Authorization: Bearer $(gcloud auth print-access-token)" \
    -d '{"locationUri": "gs://GCS_WAREHOUSE_PATH", "description": "DESCRIPTION"}' \
    -H "Content-Type: application/json" \
    "https://biglake.googleapis.com/hive/v1/projects/PROJECT_ID/catalogs?hiveCatalogId=LAKEHOUSE_CATALOG_ID&primary_location=REGION"

Ersetzen Sie Folgendes:

  • LAKEHOUSE_CATALOG_ID: der Name des Hive-Katalogs.

  • GCS_WAREHOUSE_PATH: der Cloud Storage-Pfad, in dem Ihr Hive-Warehouse gespeichert ist.

  • PROJECT_ID: Projekt-ID in Google Cloud .

  • REGION: die primäre Region des Metastores. Bei Buckets mit einer einzelnen Region sollte er mit der Bucket-Region übereinstimmen. Bei Dual-Region- oder Multiregion-Buckets sollte es sich um eine der zugehörigen Regionen handeln, in der sich das primäre Replikat des Metastores befinden soll. Die andere Region wird zum sekundären Replikat.

    Wenn Sie Hive-Kataloganfragen direkt an eine bestimmte Region weiterleiten möchten, um die Anforderungen an Datenstandort und Datenhoheit zu erfüllen, können Sie regionale Endpunkte (https://biglake.LOCATION.rep.googleapis.com) verwenden.

  • DESCRIPTION: Eine Beschreibung des Katalogs.

Spark und Hive konfigurieren und verwenden

Wenn Sie den Lakehouse-Laufzeitkatalog verwenden möchten, müssen Sie Ihre Spark-Sitzung mit bestimmten Attributen konfigurieren. Sie können diese Attribute beim Erstellen eines Managed Service for Apache Spark-Clusters festlegen oder sie bei jeder Erstellung einer Sitzung angeben.

Spark-Konfigurationsattribute

Konfigurieren Sie Ihre Spark-Sitzung oder Jobübermittlung mit den folgenden Attributen, um eine Verbindung zum Apache Hive-Katalogendpunkt herzustellen:

Attribut Wert Beschreibung
spark.hive.metastore.client.factory.class com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory Konfiguriert Spark so, dass die Hive-Client-Factory des Lakehouse-Laufzeitkatalogs verwendet wird.
spark.hive.metastore.blms.project.id PROJECT_ID Gibt die Google Cloud Projekt-ID an, die den Hive-Katalog enthält.
spark.hive.metastore.blms.catalog.default LAKEHOUSE_CATALOG_ID Gibt die Standard-Hive-Katalog-ID im Lakehouse-Laufzeitkatalog an.
spark.hive.metastore.warehouse.dir gs://GCS_WAREHOUSE_PATH Gibt den Cloud Storage-URI für das Hive-Warehouse-Verzeichnis an.
spark.sql.catalogImplementation hive Aktiviert die Unterstützung des Hive-Katalogs in der Spark-Sitzung.
spark.sql.addPartitionInBatch.size 900 Optional: Legt die maximale Batchgröße für Partitionserfassungsvorgänge fest. Weitere Informationen finden Sie unter Partitions-Batchgröße.
spark.sql.extensions io.delta.sql.DeltaSparkSessionExtension Optional: Aktiviert Delta Lake-SQL-Erweiterungen, wenn Delta-Tabellen mit einem Hive-Katalog verwendet werden.
spark.sql.catalog.spark_catalog org.apache.spark.sql.delta.catalog.DeltaCatalog Optional: Konfiguriert die Delta-Katalogimplementierung, wenn Delta-Tabellen mit einem Hive-Katalog verwendet werden.

Ersetzen Sie Folgendes:

  • PROJECT_ID: Projekt-ID in Google Cloud .

  • LAKEHOUSE_CATALOG_ID: der Name des Hive-Katalogs.

  • GCS_WAREHOUSE_PATH: der Cloud Storage-Pfad, in dem Ihr Hive-Warehouse gespeichert ist.

Spark- und Hive-Abfragen ausführen

Nachdem die Sitzung eingerichtet wurde, können Sie Vorgänge wie das Auflisten vorhandener Datenbanken, das Erstellen von Datenbanken, das Definieren von Tabellen und das Einfügen von Daten ausführen:

spark-sql

  1. Stellen Sie eine SSH-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 Attributen aus, um eine interaktive Spark SQL-Sitzung zu starten:

    spark-sql \
    --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. Nachdem die Sitzung gestartet wurde, stellt Spark eine Verbindung zum Lakehouse-Laufzeitkatalog her.

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

    
-- Show all the databases in the current project.
SHOW DATABASES;

-- Create a database.
CREATE DATABASE spark_blms_database;

-- Create a Parquet datasource table.
CREATE TABLE spark_blms_database.parquet_quick_start (id INT, name STRING) USING PARQUET;

-- Insert data into the table.
INSERT INTO TABLE spark_blms_database.parquet_quick_start VALUES (1, 'my-first-user');

    Ersetzen Sie Folgendes:

    • PROJECT_ID: Ihre Projekt-ID in Google Cloud .

    • LAKEHOUSE_CATALOG_ID: der Name des Hive-Katalogs.

    • GCS_WAREHOUSE_PATH: der Cloud Storage-Pfad, in dem Ihr Hive-Warehouse gespeichert ist.

Jupyter-Notebook

  1. Folgen Sie der Anleitung zum Ausführen eines Jupyter-Notebooks in einem Managed Service for Apache Spark-Cluster.
  2. Rufen Sie die Jupyter-Weboberfläche über den Tab Weboberflächen auf der Detailseite des Clusters in der Google Cloud Console auf.

  3. Erstellen Sie in einem neuen Notebook eine Spark-Sitzung und führen Sie dann die folgenden 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 tutorial")\
    .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_blms_db")

# Create a Parquet datasource table.
spark.sql("CREATE TABLE jupyter_blms_db.parquet_table(id INT, name STRING) USING PARQUET")

# Insert data into the table.
spark.sql("INSERT INTO TABLE jupyter_blms_db.parquet_table VALUES (1, 'my-first-user');")

# Query from table.
spark.sql("SELECT * FROM jupyter_blms_db.parquet_table;").show()

    Ersetzen Sie Folgendes:

    • PROJECT_ID: Ihre Projekt-ID in Google Cloud .

    • GCS_WAREHOUSE_PATH: der Cloud Storage-Pfad, in dem Ihr Hive-Warehouse gespeichert ist.

    • 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 und die Katalogattribute der Lakehouse-Laufzeit:

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

    # Create a database
spark.sql("CREATE DATABASE bq_spark_blms_database;")

# Create a parquet datasource table
spark.sql("CREATE TABLE bq_spark_blms_database.parquet_quick_start (id INT, name STRING) USING PARQUET;")

# Insert data into the table
spark.sql("INSERT INTO TABLE bq_spark_blms_database.parquet_quick_start VALUES (1, 'my-first-user');")

# Query from table
spark.sql("select * from bq_spark_blms_database.parquet_quick_start;").show()

    Ersetzen Sie Folgendes:

    • PROJECT_ID: Ihre Projekt-ID in Google Cloud .

    • GCS_WAREHOUSE_PATH: der Cloud Storage-Pfad, in dem Ihr Hive-Warehouse gespeichert ist.

    • LAKEHOUSE_CATALOG_ID: der Name des Hive-Katalogs.

Hive-Befehlszeile

Wenn Sie die Hive-Befehlszeile verwenden möchten, müssen Sie sie so konfigurieren, dass eine Verbindung zum Metastore hergestellt wird. Dafür stehen Ihnen zwei Möglichkeiten zur Verfügung:

  • Option 1: Dauerhafte Konfiguration. Aktualisieren Sie die Datei /etc/hive/conf/hive-site.xml auf dem primären Knoten.
  • Option 2: Temporäre Konfiguration Geben Sie beim Starten der Befehlszeile Konfigurations-Flags an.

So konfigurieren Sie die Befehlszeile und führen Ihre Abfragesitzung aus:

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

  2. Führen Sie je nach Konfigurationsmethode einen der folgenden Schritte aus:

    • So konfigurieren Sie die Einstellung dauerhaft (Option 1):

      1. Öffnen Sie /etc/hive/conf/hive-site.xml auf dem primären Knoten und fügen Sie die folgenden Attribute hinzu:

        <property>
    <name>hive.metastore.blms.project.id</name>
    <value>PROJECT_ID</value>
    <description></description>
</property>

<property>
    <name>hive.metastore.client.factory.class</name>
    <value>com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory</value>
    <description></description>
</property>
<property>
    <name>hive.metastore.blms.catalog.default</name>
    <value>LAKEHOUSE_CATALOG_ID</value>
    <description></description>
</property>
<property>
    <name>hive.metastore.warehouse.dir</name>
    <value>gs://GCS_WAREHOUSE_PATH</value>
    <description></description>
</property>

      2. Starten Sie die Hive-Befehlszeile:

        hive

    • Temporäre Konfiguration beim Start (Option 2): Starten Sie die Hive-Befehlszeile mit Konfigurations-Flags:

      hive \
--hiveconf hive.metastore.blms.project.id="PROJECT_ID" \
--hiveconf hive.metastore.blms.catalog.default="LAKEHOUSE_CATALOG_ID" \
--hiveconf hive.metastore.client.factory.class=com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory \
--hiveconf hive.metastore.warehouse.dir="gs://GCS_WAREHOUSE_PATH"

  3. Führen Sie dann die folgenden Befehle in der Hive-CLI-Sitzung aus:

    show databases;
create database hive_query_test;
create table hive_query_test.parquet_table (id INT, name STRING) stored as PARQUET;
select * from hive_query_test.parquet_table;

    Ersetzen Sie Folgendes:

    • PROJECT_ID: Ihre Projekt-ID in Google Cloud .

    • LAKEHOUSE_CATALOG_ID: der Name des Hive-Katalogs.

    • GCS_WAREHOUSE_PATH: der Cloud Storage-Pfad, in dem Ihr Hive-Warehouse gespeichert ist.

Managed Service for Apache Spark-Batchjob senden

Sie können eine PySpark-Batcharbeitslast an Managed Service for Apache Spark senden, die den Lakehouse-Laufzeitkatalog verwendet.

Dieses PySpark-Snippet initialisiert eine Spark-Sitzung, die für die Verbindung mit dem Lakehouse-Laufzeitkatalog konfiguriert ist. Dabei werden wichtige Attribute wie die Lakehouse-Client-Factory, die Projekt-ID von Google Cloud , der Standardkatalog und das Warehouse-Verzeichnis festgelegt. Nachdem die Sitzung eingerichtet wurde, wird im Code gezeigt, wie Sie vorhandene Datenbanken auflisten, eine neue Datenbank erstellen und eine Tabelle im Parquet-Format in dieser Datenbank mit Spark SQL-Befehlen definieren.

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

    from pyspark.sql import SparkSession

spark = (
    SparkSession.builder.appName("Lakehouse runtime catalog tutorial")
    .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_test")

# Create a Parquet datasource table.
spark.sql(
        "CREATE TABLE dp_serverless_test.parquet_table(id INT, name STRING) USING"
        " PARQUET"
    )

# Query from table.
spark.sql("SELECT * from  dp_serverless_test.parquet_table").show()

    Ersetzen Sie Folgendes:

    • PROJECT_ID: Projekt-ID in Google Cloud .

    • LAKEHOUSE_CATALOG_ID: der Name des Hive-Katalogs.

    • GCS_WAREHOUSE_PATH: der Cloud Storage-Pfad, in dem Ihr Hive-Warehouse gespeichert ist.

  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

    Ersetzen Sie Folgendes:

    • PYTHON_SCRIPT_FILE: Der Pfad zur PySpark-Anwendungsdatei. Dies kann ein lokaler Pfad oder der Cloud Storage-Objektpfad sein.
    • PROJECT_ID: Ihre Projekt-ID.
    • REGION: Die Region, in der der Batchjob ausgeführt werden soll.
    • CLOUD_STORAGE_BUCKET: Der Name des Cloud Storage-Buckets, der zum Staging der Abhängigkeiten einer Arbeitslast verwendet wird.

Tabelle über BigQuery abfragen

Nachdem Sie Ressourcen aus Spark im Lakehouse-Laufzeitkatalog erstellt haben, können Sie sie über BigQuery Studio abfragen.

  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: Ihr Tabellenname.