Von Dataflow verwaltete E/A für Apache Iceberg

Verwaltete E/A unterstützt die folgenden Funktionen für Apache Iceberg:

Kataloge
  • Hadoop
  • Hive
  • REST-basierte Kataloge
  • BigQuery-Metastore (erfordert Apache Beam SDK 2.62.0 oder höher, wenn der Portable Runner nicht verwendet wird)
Lesefunktionen Batchlesevorgang
Schreibfunktionen
  • Batchschreibvorgang
  • Streamingschreibvorgang
  • Dynamische Ziele
  • Dynamische Tabellenerstellung

Verwenden Sie für BigQuery-Tabellen für Apache Iceberg, verwenden Sie den BigQueryIO Connector mit der BigQuery Storage API. Die Tabelle muss bereits vorhanden sein. Die dynamische Tabellenerstellung wird nicht unterstützt.

Voraussetzungen

Die folgenden SDKs unterstützen verwaltete E/A für Apache Iceberg:

  • Apache Beam SDK für Java, Version 2.58.0 oder höher
  • Apache Beam SDK für Python, Version 2.61.0 oder höher

Konfiguration

Verwaltete E/A für Apache Iceberg unterstützt die folgenden Konfigurationsparameter:

ICEBERG-Lesevorgang

Konfiguration Typ Beschreibung
table str ID der Iceberg-Tabelle.
catalog_name str Name des Katalogs, der die Tabelle enthält.
catalog_properties map[str, str] Attribute, die zum Einrichten des Iceberg-Katalogs verwendet werden.
config_properties map[str, str] Attribute, die an die Hadoop-Konfiguration übergeben werden.
drop list[str] Eine Teilmenge von Spaltennamen, die beim Lesen ausgeschlossen werden sollen. Wenn null oder leer, werden alle Spalten gelesen.
filter str SQL-ähnliches Prädikat zum Filtern von Daten zur Scanzeit. Beispiel: „id > 5 AND status = 'ACTIVE'“. Verwendet die Apache Calcite-Syntax: https://calcite.apache.org/docs/reference.html
keep list[str] Eine Teilmenge von Spaltennamen, die ausschließlich gelesen werden sollen. Wenn null oder leer, werden alle Spalten gelesen.

ICEBERG-Schreibvorgang

Konfiguration Typ Beschreibung
table str Eine voll qualifizierte Tabellen-ID. Sie können auch eine Vorlage zum Schreiben in mehrere dynamische Ziele angeben, z. B. `dataset.my_{col1}_{col2.nested}_table`.
autosharding boolean Aktiviert das dynamische Sharding, um die Anzahl der parallelen Writer automatisch an das Datenvolumen anzupassen. Es verarbeitet Datenabweichungen, indem Partitionen weiter in mehrere Shards unterteilt werden, um Engpässe bei Schreibvorgängen mit hohem Durchsatz zu vermeiden. Nur mit dem Verteilungsmodus „hash“ verfügbar.
catalog_name str Name des Katalogs, der die Tabelle enthält.
catalog_properties map[str, str] Attribute, die zum Einrichten des Iceberg-Katalogs verwendet werden.
config_properties map[str, str] Attribute, die an die Hadoop-Konfiguration übergeben werden.
direct_write_byte_limit int32 Legt für eine Streamingpipeline das Limit für das Verschieben von Bundles in den direkten Schreibpfad fest.
distribution_mode str Definiert die Verteilung der Schreibdaten. Unterstützte Verteilungen: - none: Zeilen nicht mischen (Standard) - hash: Zeilen nach Partitionsschlüssel mischen, bevor Daten geschrieben werden
drop list[str] Eine Liste von Feldnamen, die vor dem Schreiben aus dem Eingabedatensatz gelöscht werden sollen. Kann nicht zusammen mit „keep“ und „only“ verwendet werden.
keep list[str] Eine Liste der Feldnamen, die im Eingabedatensatz beibehalten werden sollen. Alle anderen Felder werden vor dem Schreiben gelöscht. Kann nicht zusammen mit „drop“ und „only“ verwendet werden.
only str Der Name eines einzelnen Datensatzfelds, in das geschrieben werden soll. Kann nicht zusammen mit „keep“ und „drop“ verwendet werden.
partition_fields list[str] Felder, mit der eine Partitionsspezifikation erstellt wird, die beim Erstellen von Tabellen angewendet wird. Für ein Feld „foo“ sind folgende Partitionstransformationen verfügbar:
  • foo
  • truncate(foo, N)
  • bucket(foo, N)
  • hour(foo)
  • day(foo)
  • month(foo)
  • year(foo)
  • void(foo)

Weitere Informationen zu Partitionstransformationen finden Sie unter https://iceberg.apache.org/spec/#partition-transforms.

sort_fields list[str] Felder, die zum Festlegen der Sortierreihenfolge der Tabelle verwendet werden und beim Erstellen der Tabelle angewendet werden. Jeder Eintrag hat das Format <term> [asc|desc] [nulls first|nulls last], wobei <term> ein Feldname oder eine der Partitionstransformationen ist (z. B. bucket(col, 4), day(ts)). Die Richtung ist standardmäßig auf aufsteigend festgelegt. Die Reihenfolge von Nullwerten ist standardmäßig auf „nulls first“ für aufsteigend und „nulls last“ für absteigend festgelegt. Hinweis: Dadurch wird die deklarierte Sortierreihenfolge der Tabelle als Metadaten festgelegt. Beam sortiert Datensätze nicht physisch vor dem Schreiben. Weitere Informationen zu Sortierreihenfolgen finden Sie unter https://iceberg.apache.org/spec/#sort-orders.
table_properties map[str, str] Iceberg-Tabellenattribute, die beim Erstellen der Tabelle festgelegt werden sollen. Weitere Informationen zu Tabellenattributen finden Sie unter https://iceberg.apache.org/docs/latest/configuration/#table-properties.
triggering_frequency_seconds int32 Legt für eine Streamingpipeline die Häufigkeit fest, mit der Snapshots erstellt werden.

Nächste Schritte

Weitere Informationen und Codebeispiele finden Sie unter den folgenden Themen: