BigQuery-Metastore mit Dataproc verwenden

In diesem Dokument wird beschrieben, wie Sie den BigQuery-Metastore mit Dataproc in der Compute Engine verwenden. Diese Verbindung bietet einen einzigen freigegebenen Metastore, der für Open-Source-Software-Engines wie Apache Spark oder Apache Flink funktioniert.

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 die Berechtigungen zu erhalten, die Sie zum Verwenden von Spark oder Flink und Dataproc mit dem BigQuery-Metastore als Metadatenspeicher benötigen:

• Dataproc-Cluster erstellen: Dataproc-Worker (roles/dataproc.worker) im Standarddienstkonto der Compute Engine im Projekt
• BigQuery-Metastore-Tabellen in Spark oder Flink erstellen:
  • Dataproc-Worker (roles/dataproc.worker) für das Dataproc-VM-Dienstkonto im Projekt
  • BigQuery-Datenbearbeiter (roles/bigquery.dataEditor) für das Dataproc-VM-Dienstkonto im Projekt
  • Storage-Objekt-Administrator (roles/storage.objectAdmin) für das Dataproc-VM-Dienstkonto im Projekt
• BigQuery-Metastore-Tabellen in BigQuery abfragen:
  • BigQuery Data Viewer (roles/bigquery.dataViewer) für das Projekt
  • BigQuery-Nutzer (roles/bigquery.user) für das Projekt
  • Storage-Objekt-Betrachter (roles/storage.objectViewer) für das Projekt

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.

Allgemeiner Workflow

So verwenden Sie Dataproc in der Compute Engine mit dem BigQuery-Metastore:

1. Erstellen Sie einen Dataproc-Cluster oder konfigurieren Sie einen vorhandenen Cluster.
2. Stellen Sie eine Verbindung zu Ihrer bevorzugten Open-Source-Software-Engine wie Spark oder Flink her.
3. Verwenden Sie eine JAR-Datei, um das Apache Iceberg-Katalog-Plug-in auf dem Cluster zu installieren.
4. Erstellen und verwalten Sie Ihre BigQuery-Metastore-Ressourcen nach Bedarf, je nachdem, welche Open-Source-Software-Engine Sie verwenden.
5. In BigQuery auf Ihre BigQuery Metastore-Ressourcen zugreifen und sie verwenden.

BigQuery-Metastore mit Spark verbinden

In der folgenden Anleitung wird beschrieben, wie Sie Dataproc mit dem BigQuery-Metastore über interaktive Spark SQL-Abfragen verbinden.

Iceberg-Katalog-Plug-in herunterladen

Wenn Sie den BigQuery-Metastore mit Dataproc und Spark verbinden möchten, müssen Sie die JAR-Datei des BigQuery-Metastore-Iceberg-Katalog-Plug-ins verwenden.

Diese Datei ist standardmäßig in der Dataproc-Image-Version 2.2 enthalten. Wenn Ihre Dataproc-Cluster keinen direkten Zugriff auf das Internet haben, müssen Sie das Plug-in herunterladen und in einen Cloud Storage-Bucket hochladen, auf den Ihr Dataproc-Cluster zugreifen kann.

Iceberg-Katalog-Plug-in für den BigQuery-Metastore herunterladen

Dataproc-Cluster konfigurieren

Bevor Sie eine Verbindung zum BigQuery-Metastore herstellen können, müssen Sie einen Dataproc-Cluster einrichten.

Dazu können Sie einen neuen Cluster erstellen oder einen vorhandenen verwenden. Anschließend können Sie diesen Cluster verwenden, um interaktive Spark SQL-Abfragen auszuführen und Ihre BigQuery-Metastore-Ressourcen zu verwalten.

• Für das Subnetz in der Region, in der der Cluster erstellt wird, muss der private Google-Zugriff (Private Google Access, PGA) aktiviert sein. Standardmäßig haben Dataproc-Cluster-VMs, die mit der Image-Version 2.2 (Standard) oder höher erstellt wurden, nur interne IP-Adressen. Damit Cluster-VMs mit Google APIs kommunizieren können, müssen Sie den privaten Google-Zugriff im Netzwerksubnetz default (oder dem vom Nutzer angegebenen Netzwerknamen, falls zutreffend) in der Region aktivieren, in der der Cluster erstellt wird.

• Wenn Sie das Beispiel für die Zeppelin-Weboberfläche in dieser Anleitung ausführen möchten, müssen Sie einen Dataproc-Cluster mit aktivierter optionaler Zeppelin-Komponente verwenden oder erstellen.

gcloud dataproc clusters create CLUSTER_NAME \
  --project=PROJECT_ID \
  --region=LOCATION \
  --optional-components=ZEPPELIN \
  --enable-component-gateway \
  --single-node

org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.1

Spark-Job senden

Sie haben folgende Möglichkeiten, einen Spark-Job einzureichen:

gcloud dataproc jobs submit spark-sql \
--project=PROJECT_ID \
--cluster=CLUSTER_NAME \
--region==REGION \
--jars=https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.6.1/iceberg-spark-runtime-3.5_2.12-1.6.1.jar,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.6.1-1.0.1-beta.jar \
--properties=spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog, \
spark.sql.catalog.CATALOG_NAME.catalog-impl=org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog, \
spark.sql.catalog.CATALOG_NAME.gcp_project=PROJECT_ID, \
spark.sql.catalog.CATALOG_NAME.gcp_location=LOCATION, \
spark.sql.catalog.CATALOG_NAME.warehouse=WAREHOUSE_DIRECTORY \
--execute="SPARK_SQL_COMMAND"

BigQuery-Metastore mit Flink verbinden

In der folgenden Anleitung wird beschrieben, wie Sie mit dem Flink SQL-Client eine Verbindung zwischen Dataproc und dem BigQuery-Metastore herstellen.

Katalog-Plug-in installieren und eine Verbindung zu einer Flink-Sitzung herstellen

So stellen Sie eine Verbindung zwischen dem BigQuery-Metastore und Flink her:

1. Erstellen Sie einen Dataproc-Cluster mit aktivierter optionaler Flink-Komponente und verwenden Sie Dataproc 2.2 oder höher.

2. Rufen Sie in der Google Cloud -Console die Seite VM-Instanzen auf:

   Zu "VM-Instanzen"

3. Klicken Sie in der Liste der VM-Instanzen auf SSH, um eine Verbindung zu einer Dataproc-VM-Instanz herzustellen.

4. Konfigurieren Sie das Iceberg-Plug-in für benutzerdefinierte Kataloge für den BigQuery-Metastore:

   FLINK_VERSION=1.17
   ICEBERG_VERSION=1.5.2
   
   cd /usr/lib/flink
   
   sudo wget -c https://repo.maven.apache.org/maven2/org/apache/iceberg/iceberg-flink-runtime-${FLINK_VERSION}/${ICEBERG_VERSION}/iceberg-flink-runtime-${FLINK_VERSION}-${ICEBERG_VERSION}.jar -P lib
   
   sudo gcloud storage cp gs://spark-lib/bigquery/iceberg-bigquery-catalog-${ICEBERG_VERSION}-1.0.1-beta.jar lib/

5. Starten Sie die Flink-Sitzung auf YARN:

   HADOOP_CLASSPATH=`hadoop classpath`
   
   sudo bin/yarn-session.sh -nm flink-dataproc -d
   
   sudo bin/sql-client.sh embedded \
     -s yarn-session

6. So erstellst du einen Katalog in Flink:

   CREATE CATALOG CATALOG_NAME WITH (
     'type'='iceberg',
     'warehouse'='WAREHOUSE_DIRECTORY',
     'catalog-impl'='org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog',
     'gcp_project'='PROJECT_ID',
     'gcp_location'='LOCATION'
   );

   Ersetzen Sie Folgendes:

   • CATALOG_NAME: die Flink-Katalog-ID, die mit einem BigQuery-Metastore-Katalog verknüpft ist.
   • WAREHOUSE_DIRECTORY: Der Basispfad für das Warehouse-Verzeichnis (der Cloud Storage-Ordner, in dem Flink Dateien erstellt). Dieser Wert beginnt mit gs://.
   • PROJECT_ID: Die Projekt-ID des BigQuery-Metastore-Katalogs, mit dem der Flink-Katalog verknüpft ist.
   • LOCATION: den Speicherort der BigQuery-Ressourcen.

Ihre Flink-Sitzung ist jetzt mit dem BigQuery-Metastore verbunden und Sie können Flink-SQL-Befehle ausführen.

BigQuery-Metastore-Ressourcen verwalten

Nachdem Sie eine Verbindung zum BigQuery-Metastore hergestellt haben, können Sie Ressourcen basierend auf den im BigQuery-Metastore gespeicherten Metadaten erstellen und aufrufen.

Führen Sie beispielsweise die folgenden Befehle in Ihrer interaktiven Flink SQL-Sitzung aus, um eine Iceberg-Datenbank und ‑Tabelle zu erstellen.

1. Benutzerdefinierten Iceberg-Katalog verwenden:

   USE CATALOG CATALOG_NAME;

   Ersetzen Sie CATALOG_NAME durch Ihre Flink-Katalog-ID.

2. So erstellen Sie eine Datenbank, die ein Dataset in BigQuery erstellt:

   CREATE DATABASE IF NOT EXISTS DATABASE_NAME;

   Ersetzen Sie DATABASE_NAME durch den Namen Ihrer neuen Datenbank.

3. Verwenden Sie die von Ihnen erstellte Datenbank:

   USE DATABASE_NAME;

4. Erstellen Sie eine Iceberg-Tabelle. Im Folgenden wird eine Beispieltabelle für Verkäufe erstellt:

   CREATE TABLE IF NOT EXISTS ICEBERG_TABLE_NAME (
       order_number BIGINT,
       price        DECIMAL(32,2),
       buyer        ROW<first_name STRING, last_name STRING>,
       order_time   TIMESTAMP(3)
   );

   Ersetzen Sie ICEBERG_TABLE_NAME durch einen Namen für die neue Tabelle.

5. So rufen Sie Tabellenmetadaten auf:

   DESCRIBE EXTENDED ICEBERG_TABLE_NAME;

6. Tabellen in der Datenbank auflisten:

   SHOW TABLES;

Daten in die Tabelle aufnehmen

Nachdem Sie im vorherigen Abschnitt eine Iceberg-Tabelle erstellt haben, können Sie Flink DataGen als Datenquelle verwenden, um Echtzeitdaten in Ihre Tabelle aufzunehmen. Die folgenden Schritte sind ein Beispiel für diesen Workflow:

1. So erstellen Sie eine temporäre Tabelle mit DataGen:

   CREATE TEMPORARY TABLE DATABASE_NAME.TEMP_TABLE_NAME
   WITH (
       'connector' = 'datagen',
       'rows-per-second' = '10',
       'fields.order_number.kind' = 'sequence',
       'fields.order_number.start' = '1',
       'fields.order_number.end' = '1000000',
       'fields.price.min' = '0',
       'fields.price.max' = '10000',
       'fields.buyer.first_name.length' = '10',
       'fields.buyer.last_name.length' = '10'
   )
   LIKE DATABASE_NAME.ICEBERG_TABLE_NAME (EXCLUDING ALL);

   Ersetzen Sie Folgendes:

   • DATABASE_NAME: der Name der Datenbank, in der die temporäre Tabelle gespeichert werden soll.
   • TEMP_TABLE_NAME: Name der temporären Tabelle.
   • ICEBERG_TABLE_NAME: der Name der Iceberg-Tabelle, die Sie im vorherigen Abschnitt erstellt haben.

2. Legen Sie die Parallelität auf „1“ fest:

   SET 'parallelism.default' = '1';

3. Legen Sie das Intervall für den Prüfpunkt fest:

   SET 'execution.checkpointing.interval' = '10second';

4. Prüfpunkt festlegen:

   SET 'state.checkpoints.dir' = 'hdfs:///flink/checkpoints';

5. Starten Sie den Echtzeit-Streaming-Job:

   INSERT INTO ICEBERG_TABLE_NAME SELECT * FROM TEMP_TABLE_NAME;

   Die Ausgabe sieht etwa so aus:

   [INFO] Submitting SQL update statement to the cluster...
   [INFO] SQL update statement has been successfully submitted to the cluster:
   Job ID: 0de23327237ad8a811d37748acd9c10b
   

6. So prüfen Sie den Status des Streamingjobs:

   1. Rufen Sie in der Google Cloud Console die Seite Cluster auf.

      Zu den Clustern

   2. Wählen Sie Ihren Cluster aus.

   3. Klicken Sie auf den Tab Weboberflächen.

   4. Klicken Sie auf den Link YARN ResourceManager.

   5. Suchen Sie auf der Benutzeroberfläche von YARN ResourceManager nach Ihrer Flink-Sitzung und klicken Sie unter Tracking-UI auf den Link ApplicationMaster.

   6. Prüfen Sie in der Spalte Status, ob der Jobstatus Wird ausgeführt lautet.

7. Streamingdaten im Flink SQL-Client abfragen:

   SELECT * FROM ICEBERG_TABLE_NAME
   /*+ OPTIONS('streaming'='true', 'monitor-interval'='3s')*/
   ORDER BY order_time desc
   LIMIT 20;

8. Streamingdaten in BigQuery abfragen:

   SELECT * FROM `DATABASE_NAME.ICEBERG_TABLE_NAME`
   ORDER BY order_time desc
   LIMIT 20;

9. Beenden Sie den Streaming-Job im Flink SQL-Client:

   STOP JOB 'JOB_ID';

   Ersetzen Sie JOB_ID durch die Job-ID, die in der Ausgabe angezeigt wurde, als Sie den Streamingjob erstellt haben.

Nächste Schritte

• Richten Sie optionale BigQuery-Metastore-Funktionen ein.
• Tabellen aus Spark in BigQuery ansehen und abfragen