Wenn Sie den benutzerdefinierten Apache Iceberg-Katalog für BigQuery so konfigurieren, dass er mit dem Lakehouse-Laufzeitkatalog verbunden wird, können Sie Managed Service for Apache Spark-Cluster oder Managed Service for Apache Spark verwenden, um einen einzelnen, freigegebenen Katalog zu erstellen.
Im Rahmen von Borderless Lakehouse können Sie mit dieser Einrichtung entweder einen Managed Service for Apache Spark-Cluster oder Managed Service for Apache Spark verwenden, um einen einzelnen, freigegebenen Katalog zu erstellen.
Nach der Konfiguration funktioniert diese Metadatenebene mit Open-Source-Engines wie Apache Spark und Apache Flink, um Ihre offenen Tabellenformate zu verwalten.
Hinweis
- Aktivieren Sie die Abrechnung für Ihr Google Cloud Projekt in. So prüfen Sie, ob die Abrechnung für ein Projekt aktiviert ist .
Aktivieren Sie die BigQuery API und die Managed Service for Apache Spark API.
Informationen zur Funktionsweise des Lakehouse-Laufzeitkatalogs.
Erforderliche Rollen
Bitten Sie Ihren Administrator, Ihnen die folgenden IAM-Rollen zuzuweisen, um die Berechtigungen zu erhalten, die Sie zum Konfigurieren des Lakehouse-Laufzeitkatalogs benötigen:
-
Managed Service for Apache Spark-Cluster erstellen:
Dataproc-Worker (
roles/dataproc.worker) für das Compute Engine-Standarddienstkonto im Projekt -
Lakehouse-Laufzeitkatalogtabellen erstellen:
- Dataproc-Worker (
roles/dataproc.worker) für das Managed Service for Apache Spark-VM-Dienstkonto im Projekt - BigQuery-Datenbearbeiter (
roles/bigquery.dataEditor) für das Dataproc-VM-Dienstkonto im Projekt - Storage Object User (
roles/storage.objectUser) für das Dataproc-VM-Dienstkonto im Projekt
- Dataproc-Worker (
-
Lakehouse-Laufzeitkatalogtabellen abfragen:
- BigQuery Data Viewer (
roles/bigquery.dataViewer) für das Projekt - BigQuery-Nutzer (
roles/bigquery.user) für das Projekt - Storage Object Viewer (
roles/storage.objectViewer) für das Projekt
- BigQuery Data 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.
Lakehouse-Laufzeitkatalog mit Managed Service for Apache Spark konfigurieren
Sie können den Lakehouse-Laufzeitkatalog mit Managed Service for Apache Spark entweder mit Apache Spark oder Apache Flink konfigurieren:
Apache Spark
Konfigurieren Sie einen neuen Cluster. Führen Sie den folgenden
gcloud dataproc clusters createBefehl aus, um einen neuen Managed Service for Apache Spark-Cluster zu erstellen. Er enthält die Einstellungen, die Sie zum Verwenden des Lakehouse-Laufzeitkatalogs benötigen:gcloud dataproc clusters create CLUSTER_NAME \ --project=PROJECT_ID \ --region=LOCATION \ --single-node
Ersetzen Sie Folgendes:
CLUSTER_NAME: ein Name für Ihren Managed Service for Apache Spark-Cluster.PROJECT_ID: die ID des Google Cloud Projekts in dem Sie den Cluster erstellen.LOCATION: die Compute Engine-Region, in der Sie den Cluster erstellen.
Senden Sie einen Apache Spark-Job mit einer der folgenden Methoden:
Google Cloud CLI
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,BIGLAKE_ICEBERG_CATALOG_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"
Ersetzen Sie Folgendes:
PROJECT_ID: die ID des Google Cloud Projekts das den Managed Service for Apache Spark-Cluster enthält.CLUSTER_NAME: der Name des Managed Service for Apache Spark-Clusters, den Sie zum Ausführen des Apache Spark SQL-Jobs verwenden.REGION: die Compute Engine Region, in der sich Ihr Cluster befindet.BIGLAKE_ICEBERG_CATALOG_JAR: der Cloud Storage-URI des benutzerdefinierten Apache Iceberg-Katalog-Plug-ins , das verwendet werden soll. Wählen Sie je nach Apache Iceberg -Versionsnummer eine der folgenden Optionen aus:- Apache Iceberg 1.9.1:
gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.9.1-1.0.1.jar - Apache Iceberg 1.6.1:
gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.6.1-1.0.2.jar
- Apache Iceberg 1.9.1:
LOCATION: der Standort der BigQuery-Ressourcen.CATALOG_NAME: der Name des Apache Spark-Katalogs, der mit Ihrem SQL-Job verwendet werden soll.WAREHOUSE_DIRECTORY: der Cloud Storage-Ordner, der Ihr Data Warehouse enthält. Dieser Wert beginnt mitgs://.SPARK_SQL_COMMAND: die Apache Spark SQL-Abfrage, die Sie ausführen möchten. Diese Abfrage enthält die Befehle zum Erstellen Ihrer Ressourcen. Beispielsweise zum Erstellen eines Namespace und einer Tabelle.
spark-sql-Befehlszeile
Rufen Sie in der Google Cloud Console die Seite VM-Instanzen auf.
Klicken Sie in der Zeile mit dem Namen der Haupt-VM-Instanz des Managed Service for Apache Spark-Clusters auf SSH. Der Name der Haupt-VM-Instanz besteht aus dem Clusternamen gefolgt vom Suffix
-m. Die Ausgabe sieht etwa so aus:Connected, host fingerprint: ssh-rsa ... Linux cluster-1-m 3.16.0-0.bpo.4-amd64 ... ... example-cluster@cluster-1-m:~$Führen Sie im Terminal den folgenden Initialisierungsbefehl für den Lakehouse-Laufzeitkatalog aus:
spark-sql \ --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,BIGLAKE_ICEBERG_CATALOG_JAR \ --conf spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog \ --conf spark.sql.catalog.CATALOG_NAME.catalog-impl=org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog \ --conf spark.sql.catalog.CATALOG_NAME.gcp_project=PROJECT_ID \ --conf spark.sql.catalog.CATALOG_NAME.gcp_location=LOCATION \ --conf spark.sql.catalog.CATALOG_NAME.warehouse=WAREHOUSE_DIRECTORY
Ersetzen Sie Folgendes:
BIGLAKE_ICEBERG_CATALOG_JAR: der Cloud Storage-URI des benutzerdefinierten Apache Iceberg-Katalog-Plug-ins, das verwendet werden soll. Wählen Sie je nach Apache Iceberg-Versionsnummer eine der folgenden Optionen aus:- Apache Iceberg 1.9.1:
gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.9.1-1.0.1.jar - Apache Iceberg 1.6.1:
gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.6.1-1.0.2.jar
- Apache Iceberg 1.9.1:
CATALOG_NAME: der Name des Apache Spark-Katalogs, der mit Ihrem SQL-Job verwendet wird.PROJECT_ID: die Google Cloud Projekt ID für den Lakehouse-Laufzeitkatalog, mit dem Ihr Apache Spark-Katalog verknüpft ist.LOCATION: der Google Cloud Standort für den Lakehouse-Laufzeitkatalog.WAREHOUSE_DIRECTORY: der Cloud Storage-Ordner, der Ihr Data Warehouse enthält. Dieser Wert beginnt mitgs://.
Nachdem Sie die Verbindung zum Cluster hergestellt haben, wird im Apache Spark-Terminal die Eingabeaufforderung
spark-sqlangezeigt, mit der Sie Apache Spark-Jobs senden können.spark-sql (default)>
Apache Flink
Erstellen Sie einen Managed Service for Apache Spark-Cluster, bei dem die optionale Apache Flink-Komponente aktiviert ist. Verwenden Sie Managed Service for Apache Spark
2.2oder höher.Rufen Sie in der Google Cloud Console die VM-Instanzen Seite auf.
Klicken Sie in der Liste der VM-Instanzen auf SSH , um eine Verbindung zur Haupt-VM-Instanz des Managed Service for Apache Spark-Clusters herzustellen. Sie wird als Clusternamen gefolgt vom Suffix
-maufgeführt.Konfigurieren Sie das benutzerdefinierte Apache Iceberg-Katalog-Plug-in für BigQuery für den Lakehouse-Laufzeitkatalog:
FLINK_VERSION=1.19 ICEBERG_VERSION=1.6.1 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.2.jar lib/
Starten Sie die Apache Flink-Sitzung in YARN:
HADOOP_CLASSPATH=`hadoop classpath` sudo bin/yarn-session.sh -nm flink-dataproc -d sudo bin/sql-client.sh embedded \ -s yarn-session
Erstellen Sie einen Katalog in Apache 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 Apache Flink-Katalog-ID, die mit dem Lakehouse-Laufzeitkatalog verknüpft ist.WAREHOUSE_DIRECTORY: der Basispfad für das Warehouse-Verzeichnis (der Cloud Storage-Ordner, in dem Apache Flink Dateien erstellt). Dieser Wert beginnt mitgs://.PROJECT_ID: die Projekt-ID für den Lakehouse-Laufzeitkatalog, mit dem der Apache Flink-Katalog verknüpft ist.LOCATION: der Standort der BigQuery Ressourcen.
Ihre Apache Flink-Sitzung ist jetzt mit dem Lakehouse-Laufzeitkatalog verbunden und Sie können Apache Flink SQL-Befehle ausführen.
Lakehouse-Laufzeitkatalogressourcen verwalten
Nachdem Sie eine Verbindung zum Lakehouse-Laufzeitkatalog hergestellt haben, können Sie Ressourcen basierend auf den im Lakehouse-Laufzeitkatalog gespeicherten Metadaten erstellen und ansehen.
Führen Sie beispielsweise die folgenden Befehle in Ihrer interaktiven Apache Flink SQL-Sitzung aus, um eine Apache Iceberg-Datenbank und -Tabelle zu erstellen.
Verwenden Sie den benutzerdefinierten Apache Iceberg-Katalog für BigQuery:
USE CATALOG CATALOG_NAME;
Ersetzen Sie
CATALOG_NAMEdurch Ihre Apache Flink-Katalog-ID.Erstellen Sie eine Datenbank, wodurch ein Dataset in BigQuery erstellt wird:
CREATE DATABASE IF NOT EXISTS DATABASE_NAME;
Ersetzen Sie
DATABASE_NAMEdurch den Namen Ihrer neuen Datenbank.Verwenden Sie die von Ihnen erstellte Datenbank:
USE DATABASE_NAME;
Erstellen Sie eine Apache Iceberg-Tabelle. Mit dem folgenden Befehl wird eine Beispiel-Verkaufsdatenbank 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_NAMEdurch einen Namen für Ihre neue Tabelle.Tabellenmetadaten ansehen:
DESCRIBE EXTENDED ICEBERG_TABLE_NAME;
Tabellen in der Datenbank auflisten:
SHOW TABLES;
Daten in die Tabelle aufnehmen
Nachdem Sie im vorherigen Abschnitt eine Apache Iceberg-Tabelle erstellt haben, können Sie Apache Flink DataGen als Datenquelle verwenden, um Echtzeitdaten in die Tabelle aufzunehmen. Die folgenden Schritte sind ein Beispiel für diesen Workflow:
Erstellen Sie mit DataGen eine temporäre Tabelle:
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: ein Name für Ihre temporäre Tabelle.ICEBERG_TABLE_NAME: der Name der Apache Iceberg-Tabelle, die Sie im vorherigen Abschnitt erstellt haben.
Legen Sie die Parallelität auf 1 fest:
SET 'parallelism.default' = '1';
Legen Sie das Prüfpunktintervall fest:
SET 'execution.checkpointing.interval' = '10second';
Legen Sie den Prüfpunkt fest:
SET 'state.checkpoints.dir' = 'hdfs:///flink/checkpoints';
Starten Sie den Echtzeit-Streamingjob:
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
So prüfen Sie den Status des Streamingjobs:
Rufen Sie in der Google Cloud Console die Cluster Seite auf.
Wählen Sie Ihren Cluster aus.
Klicken Sie auf den Tab Weboberflächen.
Klicken Sie auf den Link YARN ResourceManager.
Suchen Sie auf der Benutzeroberfläche von YARN ResourceManager Ihre Apache Flink-Sitzung und klicken Sie unter Tracking UI auf den Link ApplicationMaster.
Prüfen Sie in der Spalte Status, ob der Jobstatus Wird ausgeführt lautet.
Streamingdaten im Apache Flink SQL-Client abfragen:
SELECT * FROM ICEBERG_TABLE_NAME /*+ OPTIONS('streaming'='true', 'monitor-interval'='3s')*/ ORDER BY order_time desc LIMIT 20;
Streamingdaten in BigQuery abfragen:
SELECT * FROM `DATABASE_NAME.ICEBERG_TABLE_NAME` ORDER BY order_time desc LIMIT 20;
Beenden Sie den Streamingjob im Apache Flink SQL-Client:
STOP JOB 'JOB_ID';
Ersetzen Sie
JOB_IDdurch die Job-ID, die in der Ausgabe angezeigt wurde, als Sie den Streamingjob erstellt haben.
Lakehouse-Laufzeitkatalog mit Managed Service for Apache Spark konfigurieren
Sie können den Lakehouse-Laufzeitkatalog mit Managed Service for Apache Spark entweder mit Apache Spark SQL oder PySpark konfigurieren.
Apache Spark SQL
Erstellen Sie eine SQL-Datei mit den Apache Spark SQL-Befehlen, die Sie im Lakehouse-Laufzeitkatalog ausführen möchten. Mit diesem Befehl werden beispielsweise ein Namespace und eine Tabelle erstellt:
CREATE NAMESPACE `CATALOG_NAME`.NAMESPACE_NAME; CREATE TABLE `CATALOG_NAME`.NAMESPACE_NAME.TABLE_NAME (id int, data string) USING ICEBERG LOCATION 'WAREHOUSE_DIRECTORY';
Ersetzen Sie Folgendes:
CATALOG_NAME: der Katalogname, der auf Ihre Apache Spark-Tabelle verweist.NAMESPACE_NAME: der Namespace-Name, der auf Ihre Apache Spark-Tabelle verweist.TABLE_NAME: ein Tabellenname für Ihre Apache Spark-Tabelle.WAREHOUSE_DIRECTORY: der URI des Cloud Storage-Ordners, in dem Ihr Data Warehouse gespeichert ist.
Senden Sie einen Apache Spark SQL-Batchjob, indem Sie den folgenden
gcloud dataproc batches submit spark-sqlBefehl ausführen:gcloud dataproc batches submit spark-sql SQL_SCRIPT_PATH \ --project=PROJECT_ID \ --region=REGION \ --subnet=projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_NAME \ --deps-bucket=BUCKET_PATH \ --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"
Ersetzen Sie Folgendes:
SQL_SCRIPT_PATH: der Pfad zur SQL-Datei, die 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.SUBNET_NAME(optional): der Name eines VPC-Subnetzes in derREGIONdas die Anforderungen an das Sitzungssubnetz erfüllt.BUCKET_PATH: der Speicherort des Cloud Storage-Bucket, in den Arbeitslastabhängigkeiten hochgeladen werden sollen. DasWAREHOUSE_DIRECTORYbefindet sich in diesem Bucket. Dasgs://-URI-Präfix des Buckets ist nicht erforderlich. Sie können den Bucket-Pfad oder den Bucket-Namen angeben, z. B.mybucketname1.LOCATION: der Standort, an dem der Batchjob ausgeführt werden soll.
Weitere Informationen zum Senden von Apache Spark-Batchjobs, siehe Apache Spark-Batcharbeitslast ausführen.
PySpark
Erstellen Sie eine Python-Datei mit den PySpark-Befehlen, die Sie im Lakehouse-Laufzeitkatalog ausführen möchten.
Mit dem folgenden Befehl wird beispielsweise eine Apache Spark-Umgebung eingerichtet, um mit Apache Iceberg-Tabellen zu interagieren, die im Lakehouse-Laufzeitkatalog gespeichert sind. Anschließend werden ein neuer Namespace und eine Apache Iceberg-Tabelle in diesem Namespace erstellt.
from pyspark.sql import SparkSession spark = SparkSession.builder \ .appName("Lakehouse runtime catalog 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.sql("USE `CATALOG_NAME`;") spark.sql("CREATE NAMESPACE IF NOT EXISTS NAMESPACE_NAME;") spark.sql("USE NAMESPACE_NAME;") spark.sql("CREATE TABLE TABLE_NAME (id int, data string) USING ICEBERG LOCATION 'WAREHOUSE_DIRECTORY';")
Ersetzen Sie Folgendes:
PROJECT_ID: die ID des Google Cloud Projekts , in dem der Batchjob ausgeführt werden soll.LOCATION: der Standort der BigQuery-Ressourcen.CATALOG_NAME: der Katalogname, der auf Ihre Apache Spark-Tabelle verweist.TABLE_NAME: ein Tabellenname für Ihre Apache Spark-Tabelle.WAREHOUSE_DIRECTORY: der URI des Cloud Storage-Ordners, in dem Ihr Data Warehouse gespeichert ist.NAMESPACE_NAME: der Namespace-Name, der auf Ihre Apache Spark-Tabelle verweist.
Senden Sie den Batchjob mit dem folgenden
gcloud dataproc batches submit pysparkBefehl:gcloud dataproc batches submit pyspark PYTHON_SCRIPT_PATH \ --version=2.2 \ --project=PROJECT_ID \ --region=REGION \ --deps-bucket=BUCKET_PATH \ --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"
Ersetzen Sie Folgendes:
PYTHON_SCRIPT_PATH: der Pfad zum Python-Skript, 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.BUCKET_PATH: der Speicherort des Cloud Storage-Bucket, in den Arbeitslastabhängigkeiten hochgeladen werden sollen. Dasgs://-URI-Präfix des Buckets ist nicht erforderlich. Sie können den Bucket-Pfad oder den Bucket-Namen angeben, z. B.mybucketname1.
Weitere Informationen zum Senden von PySpark-Batchjobs finden Sie in der gcloud-Referenz zu PySpark.