Optionale Iceberg-Komponente in Dataproc

Sie können zusätzliche Komponenten wie Iceberg installieren, wenn Sie einen Dataproc-Cluster mit dem Feature Optionale Komponenten erstellen. Auf dieser Seite wird beschrieben, wie Sie die Iceberg-Komponente optional in einem Dataproc-Cluster installieren können.

Übersicht

Apache Iceberg ist ein offenes Tabellenformat für große analytische Datasets. Es bietet die Zuverlässigkeit und Einfachheit von SQL-Tabellen für Big Data und ermöglicht es Engines wie Spark, Trino, PrestoDB, Flink und Hive, gleichzeitig sicher mit denselben Tabellen zu arbeiten.

Wenn die Apache Iceberg-Komponente auf einem Dataproc-Cluster installiert wird, werden Iceberg-Bibliotheken installiert und Spark und Hive für die Verwendung von Iceberg im Cluster konfiguriert.

Wichtige Iceberg-Funktionen

Zu den Iceberg-Funktionen gehören:

  • Schemaentwicklung:Sie können Spalten hinzufügen, entfernen oder umbenennen, ohne die gesamte Tabelle neu zu schreiben.
  • Zeitreisen:Sie können historische Tabellen-Snapshots für Auditing- oder Rollback-Zwecke abfragen.
  • Verborgene Partitionierung:Optimieren Sie das Datenlayout für schnellere Abfragen, ohne Nutzern Partitionierungsdetails preiszugeben.
  • ACID-Transaktionen:Sorgen für Datenkonsistenz und verhindern Konflikte.

Kompatible Dataproc-Image-Versionen

Sie können die Iceberg-Komponente auf Dataproc-Clustern installieren, die mit Image-Versionen 2.2.47 und höher erstellt wurden. Die im Cluster installierte Iceberg-Version ist auf der Seite 2.2-Releaseversionen aufgeführt.

Wenn Sie einen Dataproc-Cluster mit Iceberg erstellen, werden die folgenden Spark- und Hive-Attribute für die Verwendung mit Iceberg konfiguriert.

Konfigurationsdatei Attribut Standardwert
/etc/spark/conf/spark-defaults.conf spark.sql.extensions org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
spark.driver.extraClassPath /usr/lib/iceberg/lib/iceberg-spark-runtime-spark-version_scala-version.jar
spark.executor.extraClassPath /usr/lib/iceberg/lib/iceberg-spark-runtime-spark-version_scala-version.jar
/etc/hive/conf/hive-site.xml hive.aux.jars.path file:///usr/lib/iceberg/lib/iceberg-hive-runtime.jar
iceberg.engine.hive.enabled true

Optionale Iceberg-Komponente installieren

Installieren Sie die Iceberg-Komponente, wenn Sie einen Dataproc-Cluster erstellen. Auf den Seiten der Liste der Dataproc-Cluster-Image-Versionen wird die Iceberg-Komponentenversion angezeigt, die in den neuesten Dataproc-Cluster-Image-Versionen enthalten ist.

Google Cloud console

So erstellen Sie einen Dataproc-Cluster, in dem die Iceberg-Komponente installiert wird: Google Cloud

  1. Öffnen Sie die Dataproc-Seite Cluster erstellen. Der Bereich Cluster einrichten ist ausgewählt.
  2. Wählen Sie im Abschnitt Komponenten unter Optionale Komponenten die Komponente Iceberg aus.
  3. Bestätigen oder geben Sie andere Clustereinstellungen an und klicken Sie dann auf Erstellen.

Google Cloud CLI

Verwenden Sie zum Erstellen eines Dataproc-Clusters, in dem die Iceberg-Komponente installiert wird, den Befehl gcloud dataproc clusters create mit dem Flag --optional-components.

gcloud dataproc clusters create CLUSTER_NAME \
    --region=REGION \
    --optional-components=ICEBERG \
     other flags ...

Ersetzen Sie Folgendes:

REST API

Wenn Sie einen Dataproc-Cluster erstellen möchten, in dem die optionale Iceberg-Komponente installiert wird, geben Sie Iceberg SoftwareConfig.Component als Teil einer clusters.create-Anfrage an.

Iceberg-Tabellen mit Spark und Hive verwenden

Nachdem Sie einen Dataproc-Cluster erstellt haben, auf dem die optionale Iceberg-Komponente installiert ist, können Sie mit Spark und Hive Iceberg-Tabellendaten lesen und schreiben.

Spark

Spark-Sitzung für Iceberg konfigurieren

Sie können den gcloud CLI-Befehl lokal oder die spark-shell- oder pyspark-REPLs (Read-Eval-Print Loops) verwenden, die auf dem Masterknoten des Dataproc-Clusters ausgeführt werden, um die Spark-Erweiterungen von Iceberg zu aktivieren und den Spark-Katalog für die Verwendung von Iceberg-Tabellen einzurichten.

gcloud

Führen Sie das folgende gcloud CLI-Beispiel in einem lokalen Terminalfenster oder in Cloud Shell aus, um einen Spark-Job zu senden und Spark-Attribute festzulegen, um die Spark-Sitzung für Iceberg zu konfigurieren.

gcloud dataproc jobs submit spark  \
    --cluster=CLUSTER_NAME \
    --region=REGION \
    --properties="spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions" \
    --properties="spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog" \
    --properties="spark.sql.catalog.CATALOG_NAME.type=hadoop" \
    --properties="spark.sql.catalog.CATALOG_NAME.warehouse=gs://BUCKET/FOLDER" \
     other flags ...

Ersetzen Sie Folgendes:

  • CLUSTER_NAME: Der Clustername.
  • REGION: Die Compute Engine-Region.
  • CATALOG_NAME: Name des Iceberg-Katalogs.
  • BUCKET und FOLDER: Der Speicherort des Iceberg-Katalogs in Cloud Storage.

spark-shell

Führen Sie die folgenden Schritte aus, um eine Spark-Sitzung für Iceberg mit der spark-shell REPL im Dataproc-Cluster zu konfigurieren:

  1. Stellen Sie mit SSH eine Verbindung zum Masterknoten des Dataproc-Clusters her.

  2. Führen Sie im SSH-Sitzungsterminal den folgenden Befehl aus, um die Spark-Sitzung für Iceberg zu konfigurieren.

spark-shell \
    --conf "spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions" \
    --conf "spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog" \
    --conf "spark.sql.catalog.CATALOG_NAME.type=hadoop" \
    --conf "spark.sql.catalog.CATALOG_NAME.warehouse=gs://BUCKET/FOLDER"

Ersetzen Sie Folgendes:

  • CLUSTER_NAME: Der Clustername.
  • REGION: Die Compute Engine-Region.
  • CATALOG_NAME: Name des Iceberg-Katalogs.
  • BUCKET und FOLDER: Der Speicherort des Iceberg-Katalogs in Cloud Storage.

pyspark-Shell

Führen Sie die folgenden Schritte aus, um eine Spark-Sitzung für Iceberg mit der pyspark REPL im Dataproc-Cluster zu konfigurieren:

  1. Stellen Sie mit SSH eine Verbindung zum Masterknoten des Dataproc-Clusters her.

  2. Führen Sie im Terminal der SSH-Sitzung den folgenden Befehl aus, um die Spark-Sitzung für Iceberg zu konfigurieren:

pyspark \
    --conf "spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions" \
    --conf "spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog" \
    --conf "spark.sql.catalog.CATALOG_NAME.type=hadoop" \
    --conf "spark.sql.catalog.CATALOG_NAME.warehouse=gs://BUCKET/FOLDER"

Ersetzen Sie Folgendes:

  • CLUSTER_NAME: Der Clustername.
  • REGION: Die Compute Engine-Region.
  • CATALOG_NAME: Name des Iceberg-Katalogs.
  • BUCKET und FOLDER: Der Speicherort des Iceberg-Katalogs in Cloud Storage.

Daten in eine Iceberg-Tabelle schreiben

Sie können Daten mit Spark in eine Iceberg-Tabelle schreiben. Mit den folgenden Code-Snippets wird ein DataFrame mit Beispieldaten erstellt, eine Iceberg-Tabelle in Cloud Storage erstellt und die Daten in die Iceberg-Tabelle geschrieben.

PySpark

# Create a DataFrame with sample data.
data = spark.createDataFrame([(1, "Alice"), (2, "Bob")], ["id", "name"])

# Create an Iceberg table in Cloud Storage.
spark.sql("""CREATE TABLE IF NOT EXISTS CATALOG_NAME.NAMESPACE.TABLE_NAME (
    id integer,
    name string)
USING iceberg
LOCATION 'gs://BUCKET/FOLDER/NAMESPACE/TABLE_NAME'""")

# Write the DataFrame to the Iceberg table in Cloud Storage.
data.writeTo("CATALOG_NAME.NAMESPACE.TABLE_NAME").append()

Scala

// Create a DataFrame with sample data.
val data = Seq((1, "Alice"), (2, "Bob")).toDF("id", "name")

// Create an Iceberg table in Cloud Storage.
spark.sql("""CREATE TABLE IF NOT EXISTS CATALOG_NAME.NAMESPACE.TABLE_NAME (
    id integer,
    name string)
USING iceberg
LOCATION 'gs://BUCKET/FOLDER/NAMESPACE/TABLE_NAME'""")

// Write the DataFrame to the Iceberg table in Cloud Storage.
data.writeTo("CATALOG_NAME.NAMESPACE.TABLE_NAME").append()

Daten aus einer Iceberg-Tabelle lesen

Sie können Daten aus einer Iceberg-Tabelle mit Spark lesen. In den folgenden Code-Snippets wird die Tabelle gelesen und ihr Inhalt angezeigt.

PySpark

# Read Iceberg table data into a DataFrame.
df = spark.read.format("iceberg").load("CATALOG_NAME.NAMESPACE.TABLE_NAME")
# Display the data.
df.show()

Scala

// Read Iceberg table data into a DataFrame.
val df = spark.read.format("iceberg").load("CATALOG_NAME.NAMESPACE.TABLE_NAME")

// Display the data.
df.show()

Spark SQL

SELECT * FROM CATALOG_NAME.NAMESPACE.TABLE_NAME

Hive

Iceberg-Tabelle in Hive erstellen

In Dataproc-Clustern ist Hive für die Verwendung mit Iceberg vorkonfiguriert.

Führen Sie die folgenden Schritte aus, um die Code-Snippets in diesem Abschnitt auszuführen:

  1. Stellen Sie eine SSH-Verbindung zum Masterknoten Ihres Dataproc-Clusters her.

  2. Rufen Sie beeline im SSH-Terminalfenster auf.

    beeline -u jdbc:hive2://

Sie können eine nicht partitionierte oder partitionierte Iceberg-Tabelle in Hive erstellen.

Nicht partitionierte Tabelle

Erstellen Sie eine nicht partitionierte Iceberg-Tabelle in Hive.

CREATE TABLE my_table (
  id INT,
  name STRING,
  created_at TIMESTAMP
) STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler';

Partitionierte Tabelle

Erstellen Sie eine partitionierte Iceberg-Tabelle in Hive, indem Sie die Partitionsspalten in der PARTITIONED BY-Klausel angeben.

CREATE TABLE my_partitioned_table (
  id INT,
  name STRING
) PARTITIONED BY (date_sk INT)
STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler';

Daten in eine Iceberg-Tabelle in Hive einfügen

Sie können Daten mit standardmäßigen Hive-INSERT-Anweisungen in eine Iceberg-Tabelle einfügen.

SET hive.execution.engine=mr;

INSERT INTO my_table
SELECT 1, 'Alice', current_timestamp();

Beschränkungen

  • Die MR-Ausführungsengine (MapReduce) wird nur für DML-Vorgänge (Data Manipulation Language) unterstützt.
  • Die MR-Ausführung ist in Hive 3.1.3 veraltet.

Daten aus einer Iceberg-Tabelle in Hive lesen

Verwenden Sie eine SELECT-Anweisung, um Daten aus einer Iceberg-Tabelle zu lesen.

SELECT * FROM my_table;

Iceberg-Tabelle in Hive löschen

Verwenden Sie die Anweisung DROP TABLE, um eine Iceberg-Tabelle in Hive zu löschen.

DROP TABLE my_table;

Nächste Schritte