Dataflow から Apache Iceberg に書き込む

Dataflow から Apache Iceberg に書き込むには、マネージド I/O コネクタを使用します。

マネージド I/O は、Apache Iceberg の次の機能をサポートしています。

カタログ
  • Hadoop
  • Hive
  • REST ベースのカタログ
  • BigQuery metastore（Beam 2.63.0 のリリース後）
読み取り機能 バッチ読み取り
書き込み機能

Apache Iceberg 用の BigQuery テーブルの場合は、BigQuery Storage API で BigQueryIO コネクタを使用します。このテーブルはすでに存在している必要があります。動的テーブルの作成はサポートされていません。

依存関係

プロジェクトに次の依存関係を追加します。

Java

<dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-sdks-java-managed</artifactId>
  <version>${beam.version}</version>
</dependency>

<dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-sdks-java-io-iceberg</artifactId>
  <version>${beam.version}</version>
</dependency>

構成

マネージド I/O は、Apache Iceberg に次の構成パラメータを使用します。

構成の読み取りと書き込み データ型 説明
table 文字列 Apache Iceberg テーブルの識別子。例: "db.table1"。
catalog_name 文字列 カタログ名です。例: "local"。
catalog_properties マップ Apache Iceberg カタログの構成プロパティのマップ。必要なプロパティはカタログによって異なります。詳細については、Apache Iceberg ドキュメントの CatalogUtil をご覧ください。
config_properties マップ オプションの Hadoop の構成プロパティのセット。詳細については、Apache Iceberg ドキュメントの CatalogUtil をご覧ください。
構成の書き込み データ型 説明
triggering_frequency_seconds 整数 ストリーミング書き込みパイプラインの場合、シンクがスナップショットの生成を試みる頻度（秒単位）。

ダイナミック デスティネーション

Apache Iceberg のマネージド I/O は、ダイナミック デスティネーションをサポートしています。コネクタは、単一の固定テーブルに書き込む代わりに、受信レコード内のフィールド値に基づいて宛先テーブルを動的に選択できます。

ダイナミック デスティネーションを使用するには、table 構成パラメータのテンプレートを提供します。詳細については、ダイナミック デスティネーションをご覧ください。

例

次の例では、メモリ内の JSON データを Apache Iceberg テーブルに書き込みます。

Java

Dataflow への認証を行うには、アプリケーションのデフォルト認証情報を設定します。詳細については、ローカル開発環境の認証を設定するをご覧ください。

import com.google.common.collect.ImmutableMap;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.managed.Managed;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.JsonToRow;
import org.apache.beam.sdk.values.PCollectionRowTuple;

public class ApacheIcebergWrite {
  static final List<String> TABLE_ROWS = Arrays.asList(
      "{\"id\":0, \"name\":\"Alice\"}",
      "{\"id\":1, \"name\":\"Bob\"}",
      "{\"id\":2, \"name\":\"Charles\"}"
  );

  static final String CATALOG_TYPE = "hadoop";

  // The schema for the table rows.
  public static final Schema SCHEMA = new Schema.Builder()
      .addStringField("name")
      .addInt64Field("id")
      .build();

  public interface Options extends PipelineOptions {
    @Description("The URI of the Apache Iceberg warehouse location")
    String getWarehouseLocation();

    void setWarehouseLocation(String value);

    @Description("The name of the Apache Iceberg catalog")
    String getCatalogName();

    void setCatalogName(String value);

    @Description("The name of the table to write to")
    String getTableName();

    void setTableName(String value);
  }

  public static void main(String[] args) {

    // Parse the pipeline options passed into the application. Example:
    //   --runner=DirectRunner --warehouseLocation=$LOCATION --catalogName=$CATALOG \
    //   --tableName= $TABLE_NAME
    // For more information, see https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options
    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);
    Pipeline pipeline = Pipeline.create(options);

    // Configure the Iceberg source I/O
    Map catalogConfig = ImmutableMap.<String, Object>builder()
        .put("warehouse", options.getWarehouseLocation())
        .put("type", CATALOG_TYPE)
        .build();

    ImmutableMap<String, Object> config = ImmutableMap.<String, Object>builder()
        .put("table", options.getTableName())
        .put("catalog_name", options.getCatalogName())
        .put("catalog_properties", catalogConfig)
        .build();

    // Build the pipeline.
    pipeline.apply(Create.of(TABLE_ROWS))
        .apply(JsonToRow.withSchema(SCHEMA))
        .apply(Managed.write(Managed.ICEBERG).withConfig(config));

    pipeline.run().waitUntilFinish();
  }
}

次のステップ