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

背景情報

機能

制限事項

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

• Apache Iceberg コネクタは、バージョン 1 とバージョン 2 の Apache Iceberg テーブル形式をサポートしています。 詳細については、「Iceberg Table Spec」をご参照ください。

  説明

  VVR 8.0.7 以降を使用する Apache Flink 用 Realtime Compute のみ、バージョン 2 の Apache Iceberg テーブル形式をサポートしています。

• ストリーミング読み取りモードが有効になっている場合、Append Only モードでデータが書き込まれた Apache Iceberg テーブルのみをソーステーブルとして使用できます。

構文

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

WITH 句のコネクタオプション

共通オプションとソーステーブルオプション

シンク専用オプション

データ型マッピング

サンプルコード

Apache 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;

Apache Iceberg ソーステーブルの作成と使用

• Hive カタログで管理され、セルフマネージド Hive メタストアに格納されている Iceberg テーブルにマッピングする Flink テーブルを作成します。

  説明

  • Flink ワークスペースとセルフマネージド Hive メタストアクラスター間の接続が作成されていることを確認してください。

  • データファイルは oss://<bucket>/<path>/<database-name>/flink_table ディレクトリに格納されます。

  CREATE TEMOPORY TABLE flink_table (
    id   BIGINT,
    data STRING
  ) WITH (
    'connector'='iceberg',
    'catalog-name'='<yourCatalogName>',
    'catalog-database'='<yourDatabaseName>',
    'uri'='thrift://<ip>:<post>',
    'warehouse'='oss://<bucket>/<path>',
    'io-impl'='org.apache.iceberg.aliyun.oss.OSSFileIO',
    'access-key-id'='<yourAccessKeyID>',
    'access-key-secret'='<yourAccessKeySecret>',
    'oss.endpoint'='<yourOSSEndpoint>'
  );

• DLF カタログで管理されている Iceberg テーブルにマッピングする 2 つの Flink テーブルを作成し、1 つの Flink テーブルからデータを読み取ってもう 1 つのテーブルに書き込みます。

  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 Flink 用 Realtime Compute でサポートされているコネクタの詳細については、「サポートされているコネクタ」をご参照ください。