Apache Iceberg connector - Realtime Compute for Apache Flink

Topik ini menjelaskan cara menggunakan Apache Iceberg connector.

Informasi latar belakang

Apache Iceberg adalah format tabel data lake terbuka. Anda dapat menggunakan Apache Iceberg untuk dengan cepat membangun layanan penyimpanan data lake sendiri di HDFS atau Object Storage Service (OSS), serta menggunakan mesin komputasi dari ekosistem big data open-source seperti Flink, Spark, Hive, dan Presto untuk melakukan analisis data lake.

Kategori	Deskripsi
Jenis yang didukung	Tabel sumber, tabel sink, dan sink ingesti data
Mode operasional	Mode batch dan mode streaming
Format data	Tidak berlaku
Metrik spesifik	Tidak ada
Jenis API	SQL, pekerjaan ingesti data YAML
Mendukung pembaruan atau penghapusan data di tabel sink	Ya

Fitur

Apache Iceberg connector menyediakan fitur-fitur berikut:

Membangun layanan penyimpanan data lake berbiaya rendah dan ringan berbasis HDFS atau object storage.
Menyediakan semantik ACID lengkap.
Mendukung kueri time travel untuk mengakses versi data historis.
Mendukung penyaringan data yang efisien.
Mendukung evolusi skema tabel.
Mendukung evolusi partisi.

Catatan

Gunakan kemampuan toleransi kesalahan dan pemrosesan aliran Flink untuk mengimpor volume besar log ke data lake Apache Iceberg secara real time. Anda kemudian dapat menggunakan Flink atau mesin analisis lainnya untuk mengekstraksi nilai dari data tersebut.

Batasan

Apache Iceberg connector hanya didukung di Realtime Compute for Apache Flink dengan Ververica Runtime (VVR) 4.0.8 atau lebih baru. Apache Iceberg connector harus digunakan bersama katalog Data Lake Formation (DLF). Untuk informasi selengkapnya, lihat Mengelola katalog DLF-Legacy.
Apache Iceberg connector mendukung format tabel Apache Iceberg v1 dan v2. Untuk informasi selengkapnya, lihat Spesifikasi Tabel Iceberg.

Catatan
Hanya Realtime Compute for Apache Flink yang menggunakan VVR 8.0.7 atau lebih baru yang mendukung format tabel v2.
Dalam mode baca streaming, Anda hanya dapat menggunakan tabel Iceberg append-only sebagai tabel sumber.

Sintaksis

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

Opsi WITH

Opsi umum

Parameter	Deskripsi	Type	Wajib	Default	Keterangan
connector	Jenis connector.	String	Ya	None	Nilainya harus`iceberg`.
catalog-name	Nama katalog.	String	Ya	None	Masukkan nama kustom dalam bahasa Inggris.
catalog-database	Nama database.	String	Ya	default	Nama database Anda di Data Lake Formation (DLF), misalnya dlf_db. Catatan Jika Anda belum memiliki database Data Lake Formation (DLF), buat terlebih dahulu.
io-impl	Kelas implementasi untuk sistem file terdistribusi.	String	Ya	None	Nilainya harus`org.apache.iceberg.aliyun.oss.OSSFileIO`.
oss.endpoint	Titik akhir Alibaba Cloud Object Storage Service (OSS).	String	Tidak	None	Untuk informasi selengkapnya, lihat Wilayah dan Titik Akhir. Catatan Kami menyarankan Anda mengatur parameter oss.endpoint ke titik akhir VPC untuk OSS. Misalnya, jika Anda memilih wilayah China (Hangzhou), atur parameter oss.endpoint menjadi oss-cn-hangzhou-internal.aliyuncs.com. Jika Anda perlu mengakses OSS lintas VPC, lihat Bagaimana cara mengakses layanan lain lintas VPC?
`access.key.id`: untuk VVR 8.0.6 dan sebelumnya `access-key-id`: untuk VVR 8.0.7 dan seterusnya	ID AccessKey akun Alibaba Cloud Anda.	String	Ya	None	Untuk informasi selengkapnya, lihat Bagaimana cara melihat ID AccessKey dan rahasia AccessKey? Penting Untuk mencegah kebocoran informasi AccessKey Anda, kami menyarankan Anda menggunakan variabel untuk menentukan nilai AccessKey. Untuk informasi selengkapnya, lihat Variabel proyek.
`access.key.secret`: untuk VVR 8.0.6 dan sebelumnya `access-key-secret`: untuk VVR 8.0.7 dan seterusnya	Rahasia AccessKey akun Alibaba Cloud Anda.	String	Ya	None
catalog-impl	Nama kelas katalog.	String	Ya	None	Nilainya harus`org.apache.iceberg.aliyun.dlf.DlfCatalog`.
warehouse	Jalur OSS untuk menyimpan data tabel.	String	Ya	None	None
dlf.catalog-id	ID akun Alibaba Cloud Anda.	String	Ya	None	Anda dapat memperoleh ID akun di halaman Informasi Pengguna.
dlf.endpoint	Titik akhir Data Lake Formation (DLF).	String	Ya	None	. Catatan Kami menyarankan Anda mengatur parameter dlf.endpoint ke titik akhir VPC DLF. Misalnya, jika Anda memilih wilayah China (Hangzhou), atur parameter dlf.endpoint menjadi dlf-vpc.cn-hangzhou.aliyuncs.com. Jika Anda perlu mengakses DLF lintas VPC, lihat Manajemen dan operasi workspace
dlf.region-id	Wilayah Data Lake Formation (DLF).	String	Ya	None	. Catatan Pastikan wilayahnya sama dengan wilayah yang ditentukan untuk parameter dlf.endpoint.

Opsi khusus sink

Parameter	Deskripsi	Tipe	Wajib	Default	Keterangan
write.operation	Mode operasi penulisan.	String	Tidak	upsert	`upsert` (default): Memperbarui data. `insert`: Menambahkan data. `bulk_insert`: Melakukan insert massal tanpa memperbarui data yang sudah ada.
hive_sync.enable	Menentukan apakah metadata disinkronkan ke Hive.	Boolean	Tidak	false	Nilai yang valid: `true`: Mengaktifkan sinkronisasi. `false` (default): Menonaktifkan sinkronisasi.
hive_sync.mode	Mode sinkronisasi metadata Hive.	String	Tidak	hms	`hms` (default): Atur nilai ini jika Anda menggunakan katalog DLF. `jdbc`: Atur nilai ini jika Anda menggunakan katalog JDBC.
hive_sync.db	Nama database Hive tempat data disinkronkan.	String	Tidak	Nama database tabel saat ini di katalog.	None
hive_sync.table	Nama tabel Hive tempat data disinkronkan.	String	Tidak	Nama tabel saat ini.	None
dlf.catalog.region	Wilayah Data Lake Formation (DLF).	String	Tidak	None	. Catatan Parameter dlf.catalog.region hanya berlaku ketika parameter `hive_sync.mode` diatur ke`hms`. Pastikan wilayahnya sama dengan wilayah yang ditentukan untuk parameter dlf.catalog.endpoint.
dlf.catalog.endpoint	Titik akhir Data Lake Formation (DLF).	String	Tidak	None	. Catatan Parameter dlf.catalog.endpoint hanya berlaku ketika parameter `hive_sync.mode` diatur ke `hms`. Kami menyarankan Anda mengatur parameter dlf.catalog.endpoint ke titik akhir VPC DLF. Misalnya, jika Anda memilih wilayah China (Hangzhou), atur parameter dlf.catalog.endpoint menjadi dlf-vpc.cn-hangzhou.aliyuncs.com. Jika Anda perlu mengakses DLF lintas VPC, lihat Manajemen dan operasi workspace

Pemetaan tipe data

Iceberg Type	Tipe Flink
BOOLEAN	BOOLEAN
INT	INT
LONG	BIGINT
FLOAT	FLOAT
DOUBLE	DOUBLE
DECIMAL(P,S)	DECIMAL(P,S)
DATE	DATE
TIME	TIME Catatan Timestamp Iceberg memiliki presisi mikrodetik, sedangkan timestamp Flink memiliki presisi milidetik. Saat Anda menggunakan Flink untuk membaca data dari Iceberg, Flink mengonversi presisi waktu ke milidetik.
TIMESTAMP	TIMESTAMP
TIMESTAMPTZ	TIMESTAMP_LTZ
STRING	STRING
FIXED(L)	BYTES
BINARY	VARBINARY
STRUCT<...>	ROW
LIST<E>	LIST
MAP<K,V>	MAP

Contoh

Pastikan Anda memiliki bucket OSS dan database Data Lake Formation (DLF). Untuk informasi selengkapnya, lihat Membuat bucket dan Database, tabel, dan fungsi.

Catatan

Saat Anda menentukan path untuk database Data Lake Formation (DLF) Anda, kami menyarankan Anda mengikuti format ${warehouse}/${database_name}.db. Misalnya, jika alamat warehouse adalah oss://iceberg-test/warehouse dan nama database adalah dlf_db, atur jalur OSS dlf_db menjadi oss://iceberg-test/warehouse/dlf_db.db.

Contoh tabel sink

Contoh ini menggunakan connector Datagen untuk menghasilkan data streaming acak dan menulis data tersebut ke tabel 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;

Contoh tabel sumber

Gunakan katalog DLF untuk menulis data dari tabel sumber Iceberg ke tabel sink 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;

Ingesti Data

Anda dapat menggunakan Apache Iceberg connector sebagai sink dalam pekerjaan YAML untuk Ingesti Data.

Sintaksis

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

Parameter

Parameter	Deskripsi	Wajib	Tipe	Default	Keterangan
type	Jenis connector.	Ya	STRING	None	Nilai tetap adalah `iceberg`.
name	Nama sink.	Tidak	STRING	None	Nama sink.
catalog.properties.rest.signing-region	ID wilayah DLF. Untuk informasi selengkapnya, lihat Titik akhir layanan.	Ya	STRING	None	None
catalog.properties.uri	URI yang digunakan untuk mengakses katalog REST DLF. Untuk informasi selengkapnya, lihat Iceberg REST.	Ya	STRING	None	None
catalog.properties.warehouse	Direktori root untuk penyimpanan file.	Ya	STRING	None	None
catalog.properties.warehouse	Direktori root untuk penyimpanan file.	Tidak	STRING	None	None
catalog.properties.type	Jenis katalog. Nilainya harus rest.	Ya	STRING	rest	None
catalog.properties.io-impl	Nilainya harus org.apache.iceberg.rest.DlfFileIO.	Ya	STRING	org.apache.iceberg.rest.DlfFileIO	None
partition.key	Kunci partisi untuk setiap tabel partisi.	Tidak	STRING	None	Anda dapat mengatur kunci partisi untuk beberapa tabel. Pisahkan definisi tabel dengan titik koma (`;`) dan kunci partisi dengan koma (`,`). Misalnya, Anda dapat menentukan `testdb.table1:id1,id2;testdb.table2:name` untuk mengatur kunci partisi untuk tabel `testdb.table1` menjadi `id1` dan `id2`, serta kunci partisi untuk tabel `testdb.table2` menjadi `name`. Untuk partisi yang memerlukan transformasi implisit, tambahkan fungsi transformasi langsung ke bidang partisi. Contoh: `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.*	Parameter untuk membuat tabel Iceberg.	Tidak	String	None	Untuk informasi selengkapnya, lihat Opsi tabel Iceberg.

Gunakan katalog yang sudah ada

Mulai dari VVR 11.5, Anda dapat langsung mereferensikan katalog Iceberg bawaan yang dibuat di halaman Data Management dalam pekerjaan ingesti data Flink CDC. Hal ini menyederhanakan konfigurasi Anda dengan mengurangi jumlah properti koneksi yang diperlukan.

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

Pekerjaan ingesti data dapat secara otomatis menggunakan kembali semua parameter katalog Iceberg. Ini setara dengan mengonfigurasi parameter secara manual dengan awalan catalog.properties. dalam pekerjaan YAML Anda.

Jika Anda ingin mengganti parameter yang digunakan kembali, Anda dapat secara eksplisit menentukan parameter YAML yang sesuai. Parameter ini akan memiliki prioritas lebih tinggi.

Contoh

Contoh berikut menunjukkan cara menggunakan katalog DLF sebagai katalog Iceberg dan menulis data ke 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

Untuk informasi tentang parameter yang diawali dengan catalog.properties, lihat Membuat katalog Iceberg DLF.

Perubahan skema

Saat digunakan sebagai sink ingesti data, Apache Iceberg connector mendukung perubahan skema berikut:

CREATE TABLE
ADD COLUMN
ALTER COLUMN TYPE (Mengubah tipe kolom kunci primer tidak didukung.)
RENAME COLUMN
DROP COLUMN
TRUNCATE TABLE
DROP TABLE

Catatan

Jika tabel Iceberg downstream sudah ada, pekerjaan akan menggunakan skema tabel yang ada untuk penulisan dan tidak membuat ulang tabel tersebut.

Dokumentasi terkait

Untuk informasi selengkapnya tentang connector yang didukung oleh Flink, lihat Connector yang didukung.