すべてのプロダクト
Search

このプロダクト

    Realtime Compute for Apache Flink:Apache Iceberg コネクタ

    ドキュメントセンター

    Realtime Compute for Apache Flink:Apache Iceberg コネクタ

    最終更新日:Jun 04, 2026

    本トピックでは、 Apache Iceberg コネクタの使用方法を説明します。

    背景情報

    Apache Iceberg は、オープンなデータレイク向けテーブルフォーマットです。Apache Iceberg を使用すると、HDFS またはクラウド OSS 上で独自のデータレイクストレージサービスを迅速に構築でき、Flink、Spark、Hive、Presto など、オープンソースのビッグデータエコシステムのコンピューティングエンジンを使用してデータレイク分析を実行できます。

    カテゴリ

    説明

    サポートされているタイプ

    ソーステーブル、シンクテーブル、データインジェストシンク

    実行モード

    バッチモードとストリーミングモード

    データフォーマット

    該当なし

    特定のメトリクス

    なし

    API タイプ

    SQL、データインジェスト YAML ジョブ

    シンクテーブルにおけるデータの更新または削除

    はい

    機能

    Apache Iceberg コネクタは、以下の機能を提供します。

    • HDFS またはオブジェクトストレージに基づく、低コストで軽量なデータレイクストレージサービスの構築。

    • 完全な ACID セマンティクスの提供。

    • 履歴データバージョンにアクセスするためのタイムトラベルクエリのサポート。

    • 効率的なデータフィルタリングのサポート。

    • スキーマ進化のサポート。

    • パーティション進化のサポート。

    説明

    Flink のフォールトトレランスとストリーム処理機能を使用して、大量のログデータを Apache Iceberg データレイクにリアルタイムでインポートできます。その後、Flink または他の分析エンジンを使用して、データから価値を抽出できます。

    制限

    • Apache Iceberg コネクタは、Ververica Runtime (VVR) 4.0.8 以降を使用する Realtime Compute for Apache Flink でのみサポートされます。Apache Iceberg コネクタは、Data Lake Formation (DLF) カタログと併用する必要があります。詳細については、「DLF-Legacy カタログの管理」をご参照ください。

    • Apache Iceberg コネクタは、Apache Iceberg v1 および v2 テーブルフォーマットをサポートしています。詳細については、「Iceberg テーブル仕様」をご参照ください。

      説明

      v2 テーブルフォーマットは、VVR 8.0.7 以降を使用する Realtime Compute for Apache Flink でのみサポートされます。

    • ストリーミング読み取りモードでは、追加専用の Iceberg テーブルのみをソーステーブルとして使用できます。

    構文

    CREATE TABLE iceberg_table (
  id    BIGINT,
  data  STRING
  PRIMARY KEY(`id`) NOT ENFORCED
)
 PARTITIONED BY (data)
 WITH (
 'connector' = 'iceberg',
  ...
);

    WITH オプション

    共通オプション

    パラメーター

    説明

    タイプ

    必須

    デフォルト

    備考

    connector

    コネクタのタイプです。

    String

    はい

    なし

    値は iceberg である必要があります。

    catalog-name

    カタログの名前です。

    String

    はい

    なし

    英語でカスタム名を入力します。

    catalog-database

    データベースの名前です。

    String

    はい

    なし

    Data Lake Formation (DLF) のデータベース名 (例: dlf_db) です。

    説明

    Data Lake Formation (DLF) データベースがない場合は、まず作成してください。

    io-impl

    分散ファイルシステムの実装クラスです。

    String

    はい

    なし

    値は org.apache.iceberg.aliyun.oss.OSSFileIO である必要があります。

    oss.endpoint

    Alibaba Cloud Object Storage Service (OSS) のエンドポイントです。

    String

    いいえ

    なし

    詳細については、「リージョンとエンドポイント」をご参照ください。

    説明

    • oss.endpoint パラメーターを OSS の VPC エンドポイントに設定することを推奨します。 たとえば、China (Hangzhou) リージョンを選択した場合、oss.endpoint パラメーターを oss-cn-hangzhou-internal.aliyuncs.com に設定します。

    • VPC をまたいで OSS にアクセスする必要がある場合は、「How do I access other services across VPCs?」をご参照ください。

    • access.key.id： VVR 8.0.6 以前の場合

    • access-key-id： VVR 8.0.7 以降の場合

    Alibaba Cloud アカウントの AccessKey ID です。

    String

    はい

    なし

    詳細については、「How do I view the AccessKey ID and AccessKey secret?」をご参照ください。

    重要

    AccessKey 情報の漏洩を防ぐために、変数を使用して AccessKey の値を指定することを推奨します。 詳細については、「プロジェクト変数」をご参照ください。

    • access.key.secret： VVR 8.0.6 以前の場合

    • access-key-secret： VVR 8.0.7 以降の場合

    Alibaba Cloud アカウントの AccessKey シークレットです。

    String

    はい

    なし

    catalog-impl

    カタログのクラス名です。

    String

    はい

    なし

    値は org.apache.iceberg.aliyun.dlf.DlfCatalog である必要があります。

    warehouse

    テーブルデータを格納する OSS パスです。

    String

    はい

    なし

    なし

    dlf.catalog-id

    Alibaba Cloud アカウント ID です。

    String

    はい

    なし

    アカウント ID は「ユーザー情報」ページで取得できます。

    dlf.endpoint

    Data Lake Formation (DLF) のエンドポイントです。

    String

    はい

    なし

    。

    説明

    • dlf.endpoint パラメーターを DLF の VPC エンドポイントに設定することを推奨します。 たとえば、China (Hangzhou) リージョンを選択した場合、dlf.endpoint パラメーターを dlf-vpc.cn-hangzhou.aliyuncs.com に設定します。

    • VPC をまたいで DLF にアクセスする必要がある場合は、「ワークスペースの管理と操作」をご参照ください。

    dlf.region-id

    Data Lake Formation (DLF) のリージョンです。

    String

    はい

    なし

    。

    説明

    dlf.endpoint パラメーターで指定したリージョンと同じであることを確認してください。

    シンク専用オプション

    パラメーター

    説明

    タイプ

    必須

    デフォルト

    備考

    write.operation

    書き込み操作モードです。

    String

    いいえ

    upsert

    • upsert (デフォルト)： データを更新します。

    • insert： データを追記します。

    • bulk_insert： 既存のデータを更新せずに一括挿入を実行します。

    hive_sync.enable

    メタデータを Hive に同期するかどうかを指定します。

    ブール値

    いいえ

    false

    有効な値：

    • true： 同期を有効にします。

    • false (デフォルト)： 同期を無効にします。

    hive_sync.mode

    Hive メタデータの同期モードです。

    String

    いいえ

    hms

    • hms (デフォルト)： DLF カタログを使用する場合は、この値を設定します。

    • jdbc： JDBC カタログを使用する場合は、この値を設定します。

    hive_sync.db

    データの同期先となる Hive データベースの名前です。

    String

    いいえ

    カタログ内の現在のテーブルのデータベース名です。

    なし

    hive_sync.table

    データの同期先となる Hive テーブルの名前です。

    String

    いいえ

    現在のテーブルの名前です。

    なし

    dlf.catalog.region

    Data Lake Formation (DLF) のリージョンです。

    String

    いいえ

    なし

    。

    説明

    • dlf.catalog.region パラメーターは、hive_sync.mode パラメーターが hms に設定されている場合にのみ有効になります。

    • dlf.catalog.endpoint パラメーターで指定したリージョンと同じであることを確認してください。

    dlf.catalog.endpoint

    Data Lake Formation (DLF) のエンドポイントです。

    String

    いいえ

    なし

    。

    説明

    • dlf.catalog.endpoint パラメーターは、hive_sync.mode パラメーターが hms に設定されている場合にのみ有効になります。

    • dlf.catalog.endpoint パラメーターを DLF の VPC エンドポイントに設定することを推奨します。 たとえば、China (Hangzhou) リージョンを選択した場合、dlf.catalog.endpoint パラメーターを dlf-vpc.cn-hangzhou.aliyuncs.com に設定します。

    • VPC をまたいで DLF にアクセスする必要がある場合は、「ワークスペースの管理と操作」をご参照ください。

    データ型マッピング

    Iceberg 型

    Flink 型

    BOOLEAN

    BOOLEAN

    INT

    INT

    LONG

    BIGINT

    FLOAT

    FLOAT

    DOUBLE

    DOUBLE

    DECIMAL(P,S)

    DECIMAL(P,S)

    DATE

    DATE

    TIME

    TIME

    説明

    Iceberg のタイムスタンプはマイクロ秒の精度ですが、Flink のタイムスタンプはミリ秒の精度です。Flink を使用して Iceberg からデータを読み取ると、Flink によって時間の精度がミリ秒に変換されます。

    TIMESTAMP

    TIMESTAMP

    TIMESTAMPTZ

    TIMESTAMP_LTZ

    STRING

    STRING

    FIXED(L)

    BYTES

    BINARY

    VARBINARY

    STRUCT<...>

    ROW

    LIST<E>

    LIST

    MAP<K,V>

    MAP

    例

    OSS バケットと Data Lake Formation (DLF) データベースがあることを確認してください。 詳細については、「バケットの作成」および「データベース、テーブル、および関数」をご参照ください。

    説明

    Data Lake Formation (DLF) データベースの [パス] を指定する場合は、 ${warehouse}/${database_name}.db 形式を推奨します。たとえば、ウェアハウスのアドレスが oss://iceberg-test/warehouse で、データベース名が dlf_db の場合は、 dlf_db の OSS パスを oss://iceberg-test/warehouse/dlf_db.db に設定します。

    シンクテーブルの例

    この例では、Datagen コネクタを使用してランダムなストリーミングデータを生成し、そのデータを Iceberg テーブルに書き込みます。

    CREATE TEMPORARY TABLE datagen(
  id    BIGINT,
  data  STRING
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE dlf_iceberg (
  id    BIGINT,
  data  STRING
) WITH (
  'connector' = 'iceberg',
  'catalog-name' = '<yourCatalogName>',
  'catalog-database' = '<yourDatabaseName>',
  'io-impl' = 'org.apache.iceberg.aliyun.oss.OSSFileIO',
  'oss.endpoint' = '<yourOSSEndpoint>',  
  'access.key.id' = '${secret_values.ak_id}',
  'access.key.secret' = '${secret_values.ak_secret}',
  'catalog-impl' = 'org.apache.iceberg.aliyun.dlf.DlfCatalog',
  'warehouse' = '<yourOSSWarehousePath>',
  'dlf.catalog-id' = '<yourCatalogId>',
  'dlf.endpoint' = '<yourDLFEndpoint>',  
  'dlf.region-id' = '<yourDLFRegionId>'
);

INSERT INTO dlf_iceberg SELECT * FROM datagen;

    ソーステーブルの例

    • DLF カタログを使用して、Iceberg ソーステーブルのデータを Iceberg シンクテーブルに書き込みます。

      CREATE TEMPORARY TABLE src_iceberg (
  id    BIGINT,
  data  STRING
) WITH (
  'connector' = 'iceberg',
  'catalog-name' = '<yourCatalogName>',
  'catalog-database' = '<yourDatabaseName>',
  'io-impl' = 'org.apache.iceberg.aliyun.oss.OSSFileIO',
  'oss.endpoint' = '<yourOSSEndpoint>',  
  'access.key.id' = '${secret_values.ak_id}',
  'access.key.secret' = '${secret_values.ak_secret}',
  'catalog-impl' = 'org.apache.iceberg.aliyun.dlf.DlfCatalog',
  'warehouse' = '<yourOSSWarehousePath>',
  'dlf.catalog-id' = '<yourCatalogId>',
  'dlf.endpoint' = '<yourDLFEndpoint>',  
  'dlf.region-id' = '<yourDLFRegionId>'
);

CREATE TEMPORARY TABLE dst_iceberg (
  id    BIGINT,
  data  STRING
) WITH (
  'connector' = 'iceberg',
  'catalog-name' = '<yourCatalogName>',
  'catalog-database' = '<yourDatabaseName>',
  'io-impl' = 'org.apache.iceberg.aliyun.oss.OSSFileIO',
  'oss.endpoint' = '<yourOSSEndpoint>',  
  'access.key.id' = '${secret_values.ak_id}',
  'access.key.secret' = '${secret_values.ak_secret}',
  'catalog-impl' = 'org.apache.iceberg.aliyun.dlf.DlfCatalog',
  'warehouse' = '<yourOSSWarehousePath>',
  'dlf.catalog-id' = '<yourCatalogId>',
  'dlf.endpoint' = '<yourDLFEndpoint>',  
  'dlf.region-id' = '<yourDLFRegionId>'
);

BEGIN STATEMENT SET;

INSERT INTO src_iceberg VALUES (1, 'AAA'), (2, 'BBB'), (3, 'CCC'), (4, 'DDD'), (5, 'EEE');
INSERT INTO dst_iceberg SELECT * FROM src_iceberg;

END;

    データインジェスト

    Apache Iceberg コネクタは、データインジェスト用の YAML ジョブでシンクとして使用できます。

    構文

    sink:
  type: iceberg
  name: Iceberg Sink
  catalog.properties.rest.signing-region: cn-beijing
  catalog.properties.uri: http://cn-beijing-vpc.dlf.aliyuncs.com/iceberg
  catalog.properties.warehouse: flink_iceberg
  catalog.properties.type: rest
  catalog.properties.io-impl: org.apache.iceberg.rest.DlfFileIO

    パラメーター

    パラメーター

    説明

    必須

    タイプ

    デフォルト

    備考

    type

    コネクタのタイプ。

    はい

    文字列

    なし

    固定値は iceberg です。

    name

    シンクの名前。

    いいえ

    文字列

    なし

    シンクの名前。

    catalog.properties.rest.signing-region

    DLF のリージョン ID。詳細については、「サービスエンドポイント」をご参照ください。

    はい

    文字列

    なし

    なし

    catalog.properties.uri

    DLF REST カタログへのアクセスに使用する URI。詳細については、「Iceberg REST」をご参照ください。

    はい

    文字列

    なし

    なし

    catalog.properties.warehouse

    ファイルストレージのルートディレクトリ。

    はい

    文字列

    なし

    なし

    catalog.properties.warehouse

    The root directory for file storage.

    No

    STRING

    None

    None

    catalog.properties.type

    カタログのタイプ。値は rest である必要があります。

    はい

    文字列

    rest

    なし

    catalog.properties.io-impl

    値は org.apache.iceberg.rest.DlfFileIO である必要があります。

    はい

    文字列

    org.apache.iceberg.rest.DlfFileIO

    なし

    partition.key

    The partition key for each partitioned table.

    いいえ

    文字列

    なし

    複数のテーブルにパーティションキーを設定できます。テーブル定義はセミコロン (;) で、パーティションキーはコンマ (,) で区切ります。たとえば、testdb.table1:id1,id2;testdb.table2:name のように指定すると、testdb.table1 テーブルのパーティションキーが id1 および id2 に、testdb.table2 テーブルのパーティションキーが name に設定されます。

    暗黙的な変換が必要なパーティションについては、変換関数をパーティションフィールドに直接追加します。例: testdb.table1:truncate[10](id);testdb.table2:hour(create_time);testdb.table3:day(create_time);testdb.table4:month(create_time);testdb.table5:year(create_time);testdb.table6:bucket[10](create_time)。

    table.properties.*

    Iceberg テーブルを作成するためのパラメーター。

    いいえ

    文字列

    なし

    詳細については、「Iceberg テーブルオプション」をご参照ください。

    既存のカタログの再利用

    VVR 11.5 以降、Flink CDC データインジェストジョブでは、データ管理ページで作成した組み込み Iceberg カタログを直接参照できます。これにより、必要な接続プロパティの数が減り、設定が簡素化されます。

    sink:
  type: iceberg
  using.built-in-catalog: iceberg_catalog

    データ取り込みジョブは、すべての Iceberg カタログパラメーターを自動的に再利用できます。これは、YAML ジョブで catalog.properties. プレフィックスを使用して手動でパラメーターを設定することと同等です。

    再利用されたパラメーターをオーバーライドする場合は、対応する YAML パラメーターを明示的に指定できます。これらのパラメーターが優先されます。

    例

    次の例は、DLF カタログを Iceberg カタログとして使用し、Data Lake Formation (DLF) にデータを書き込む方法を示しています。

    • source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604

sink:
  type: iceberg
  name: Iceberg Sink
  catalog.properties.rest.signing-region: cn-beijing
  catalog.properties.uri: http://cn-beijing-vpc.dlf.aliyuncs.com/iceberg
  catalog.properties.warehouse: flink_iceberg
  catalog.properties.type: rest
  catalog.properties.io-impl: org.apache.iceberg.rest.DlfFileIO

      catalog.properties というプレフィックスが付いたパラメーターの詳細については、「Iceberg DLF カタログの作成」をご参照ください。

    スキーマ変更

    データインジェストシンクとして使用する場合、Apache Iceberg コネクタは以下のスキーマ変更をサポートします。

    • CREATE TABLE

    • ADD COLUMN

    • ALTER COLUMN TYPE (プライマリキー列のタイプの変更はサポートされていません。)

    • RENAME COLUMN

    • DROP COLUMN

    • TRUNCATE TABLE

    • DROP TABLE

    説明

    ダウンストリームの Iceberg テーブルが既に存在する場合、ジョブは書き込みに既存のテーブルスキーマを使用し、テーブルを再作成しません。

    関連ドキュメント

    Flink がサポートするコネクタの詳細については、「サポートされているコネクタ」をご参照ください。