Realtime Compute for Apache Flink:Apache Iceberg

背景情報

特徴

制限事項

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

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

  説明

  v2 テーブルフォーマットは、リアルタイム計算エンジン VVR 8.0.7 以降でのみサポートされています。

• ストリーム読み取りモードでは、ソーステーブルとして追加のみの Iceberg テーブルのみがサポートされます。

構文

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

WITH パラメーター

共通パラメーター (ソーステーブル用)

結果テーブル固有のパラメーター

型マッピング

コード例

結果テーブルの例

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;

ソーステーブルの例

• 自己管理型 Hive Metastore (HMS) で Hive カタログを使用する。

  Flink クラスターがネットワーク経由で HMS クラスターと通信できることを確認してください。データは 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>:<port>',
    '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 ソーステーブルから 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;

参考文献

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