Realtime Compute for Apache Flink：Konektor Apache Iceberg

Informasi latar belakang

Fitur

Batasan

• Hanya Realtime Compute for Apache Flink yang menggunakan Ververica Runtime (VVR) 4.0.8 atau lebih baru yang mendukung Konektor Apache Iceberg. Konektor Apache Iceberg harus digunakan bersama dengan katalog Data Lake Formation (DLF). Untuk informasi lebih lanjut, lihat Kelola katalog DLF.

• Konektor Apache Iceberg mendukung format tabel Apache Iceberg versi 1 dan versi 2. Untuk informasi lebih lanjut, lihat Spesifikasi Tabel Iceberg.

  Catatan

  Hanya Realtime Compute for Apache Flink yang menggunakan VVR 8.0.7 atau lebih baru yang mendukung format tabel Apache Iceberg versi 2.

• Jika mode pembacaan streaming diaktifkan, hanya tabel Apache Iceberg di mana data ditulis dalam mode Append Only yang dapat digunakan sebagai tabel sumber.

Sintaksis

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

Opsi konektor dalam klausa WITH

Opsi umum dan tabel sumber

Opsi eksklusif sink

Pemetaan tipe data

Contoh kode

Membuat dan menggunakan tabel sink 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;

Membuat dan menggunakan tabel sumber Apache Iceberg

• Buat tabel Flink yang dipetakan ke tabel Iceberg, yang dikelola dalam katalog Hive dan disimpan dalam metastore Hive yang dikelola sendiri:

  Catatan

  • Pastikan Anda telah membuat koneksi antara ruang kerja Flink dan kluster metastore Hive yang dikelola sendiri.

  • File data akan disimpan di direktori: 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>'
  );

• Buat dua tabel Flink yang dipetakan ke tabel Iceberg yang dikelola dalam katalog DLF, lalu baca data dari satu tabel Flink dan tulis ke tabel lainnya:

  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;

Referensi

Untuk informasi lebih lanjut tentang konektor yang didukung oleh Realtime Compute for Apache Flink, lihat Konektor yang Didukung.