Konektor Paimon - Realtime Compute for Apache Flink - Alibaba Cloud Documentation Center

Topik ini menjelaskan cara menggunakan konektor Paimon untuk streaming data lakehouse. Untuk hasil terbaik, kami merekomendasikan penggunaan konektor ini bersama Paimon Catalog.

Informasi latar belakang

Apache Paimon adalah format penyimpanan data lake terpadu untuk aliran (streaming) dan batch yang mendukung penulisan throughput tinggi serta kueri latensi rendah. Paimon terintegrasi dengan baik bersama mesin komputasi populer yang tersedia di Alibaba Cloud E-MapReduce, seperti Flink, Spark, Hive, dan Trino. Anda dapat menggunakan Apache Paimon untuk membangun data lake secara cepat di HDFS atau OSS dan menghubungkannya ke mesin komputasi tersebut guna melakukan analitik data lake. Untuk informasi lebih lanjut, lihat Apache Paimon.

Kategori	Deskripsi
Tipe yang didukung	Tabel sumber, tabel dimensi, tabel hasil, dan target untuk ingesti data
Mode eksekusi	Mode streaming dan mode batch
Format data	Tidak didukung
Metrik pemantauan	Tidak ada
Tipe API	SQL dan YAML untuk ingesti data
Pembaruan dan penghapusan pada tabel hasil	Ya

Fitur utama

Apache Paimon menyediakan kemampuan inti berikut:

Membangun data lake ringan dan berbiaya rendah di HDFS atau object storage.
Membaca dan menulis dataset berskala besar dalam mode streaming maupun batch.
Menjalankan kueri batch dan OLAP dengan freshness data dari hitungan menit hingga detik.
Menyuntikkan dan menghasilkan data inkremental, berfungsi sebagai lapisan penyimpanan baik untuk gudang data offline tradisional maupun gudang data streaming modern.
Melakukan pre-agregasi data untuk mengurangi biaya penyimpanan dan beban komputasi downstream.
Mengakses versi historis data.
Menyaring data secara efisien.
Mendukung evolusi skema.

Batasan dan rekomendasi

Konektor Paimon memerlukan mesin komputasi Flink VVR 6.0.6 atau versi yang lebih baru.
Tabel berikut mencantumkan kompatibilitas versi antara Paimon dan VVR.

Versi Apache Paimon

VVR

1.3.1

11.5, 11.6

1.3

11.4

1.2

11.2, 11.3

1.1

11.1

1.0

8.0.11

0.9

8.0.7, 8.0.8, 8.0.9, 8.0.10
Rekomendasi penyimpanan untuk penulisan konkuren

Saat beberapa job melakukan penulisan konkuren ke tabel Paimon yang sama, penggunaan penyimpanan OSS standar (oss://) kadang-kadang dapat menyebabkan konflik commit atau kegagalan job karena keterbatasan operasi file atomik.

Untuk penulisan yang stabil dan konsisten, gunakan layanan metadata atau penyimpanan yang menyediakan jaminan atomik kuat. Opsi yang disarankan adalah Data Lake Formation (DLF), yang menawarkan manajemen terpadu untuk metadata dan penyimpanan Paimon. Sebagai alternatif, Anda dapat menggunakan OSS-HDFS atau HDFS.
Cara perubahan konfigurasi diterapkan

Perubahan pada parameter konfigurasi tabel Paimon hanya berlaku setelah Anda me-restart job terkait. Job yang sedang berjalan tidak memuat perubahan tersebut secara dinamis.
Penghapusan fisik tertunda untuk partisi yang di-drop

Saat Anda menjalankan operasi DROP PARTITION, sistem tidak langsung menghapus file data fisik yang mendasarinya.
Operasi ini hanya melakukan penghapusan logis. Paimon hanya menghapus metadata partisi target dari snapshot terbaru. Karena Paimon mendukung fitur time travel, snapshot historis masih mereferensikan file data partisi tersebut. File data fisik tersebut baru dihapus secara permanen setelah semua snapshot historis yang mereferensikan partisi tersebut mencapai batas retensi mereka dan dibersihkan oleh mekanisme kedaluwarsa snapshot.

SQL

Gunakan konektor Paimon dalam job SQL sebagai tabel sumber atau tabel sink.

Sintaks

Jika Anda membuat tabel Paimon di Paimon Catalog, Anda tidak perlu menentukan parameter connector. Sintaksnya sebagai berikut:
```
CREATE TABLE `<YOUR-PAIMON-CATALOG>`.`<YOUR-DB>`.paimon_table (
  id BIGINT,
  data STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  ...
);
```
Catatan
Jika Anda telah membuat tabel Paimon di Paimon Catalog, Anda dapat langsung menggunakannya.
Jika Anda membuat tabel temporary Paimon di Catalog lain, Anda harus menentukan parameter 'connector' dan 'path'. Sintaksnya sebagai berikut:
```
CREATE TEMPORARY TABLE paimon_table (
  id BIGINT,
  data STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'paimon',
  'path' = '<path-to-paimon-table-files>',
  'auto-create' = 'true', -- Jika file data tabel Paimon tidak ada di path yang ditentukan, file tersebut akan dibuat secara otomatis.
  ...
);
```
Catatan
- Contoh path: 'path' = 'oss://<bucket>/test/order.db/orders'. Jangan menghilangkan akhiran .db. Paimon mengandalkan akhiran ini untuk mengidentifikasi database.
- Beberapa job yang menulis ke tabel yang sama harus menggunakan konfigurasi path yang sama.
- Jika dua konfigurasi path berbeda, Paimon tidak menganggapnya sebagai tabel yang sama. Bahkan jika path fisiknya sama, konfigurasi Catalog yang tidak konsisten dapat menyebabkan konflik penulisan konkuren, kegagalan operasi compact, dan kehilangan data. Misalnya, Paimon menganggap oss://b/test dan oss://b/test/ sebagai tabel yang berbeda karena adanya garis miring di akhir, meskipun keduanya mungkin mengarah ke lokasi fisik yang sama.

Parameter WITH

Parameter	Deskripsi	Tipe	Wajib	Default	Keterangan
connector	Menentukan konektor untuk tabel.	String	Tidak	Tidak ada	Parameter ini tidak diperlukan jika Anda membuat tabel Paimon di Paimon catalog. Jika Anda membuat tabel temporary Paimon di catalog lain, parameter ini harus diatur ke `paimon`.
path	Path penyimpanan tabel.	String	Tidak	Tidak ada	Parameter ini tidak diperlukan jika Anda membuat tabel Paimon di Paimon catalog. Jika Anda membuat tabel temporary Paimon di catalog lain, parameter ini menentukan direktori penyimpanan tabel di HDFS atau OSS.
auto-create	Menentukan apakah akan membuat file tabel secara otomatis jika file tersebut tidak ada di path yang ditentukan.	Boolean	Tidak	false	Nilai yang valid: false (default): Jika file tabel Paimon tidak ada di path yang ditentukan, job gagal. true: Jika path yang ditentukan tidak ada, Flink secara otomatis membuat file tabel Paimon.
file.format	Format file data.	String	Tidak	parquet	Nilai yang valid: orc parquet avro lance (Didukung di Realtime Compute for Apache Flink 11.6 dan versi yang lebih baru)
bucket	Jumlah bucket per partisi.	Integer	Tidak	1	Paimon mendistribusikan data ke bucket berdasarkan `bucket-key`. Catatan Kami merekomendasikan agar setiap bucket berisi kurang dari 5 GB data.
bucket-key	Kolom yang digunakan sebagai kunci bucket.	String	Tidak	Tidak ada	Menentukan kolom yang digunakan untuk mendistribusikan data ke bucket. Pisahkan beberapa nama kolom dengan koma (,). Contohnya, `'bucket-key' = 'order_id,cust_id'` mendistribusikan data berdasarkan kolom `order_id` dan `cust_id`. Catatan Jika parameter ini tidak ditentukan, Paimon mendistribusikan data berdasarkan kunci primer. Jika tabel tidak memiliki kunci primer, Paimon mendistribusikan data berdasarkan nilai semua kolom.
changelog-producer	Mekanisme produksi changelog.	String	Tidak	none	Paimon dapat menghasilkan changelog lengkap untuk aliran input apa pun, artinya setiap catatan `update_after` memiliki catatan `update_before` yang sesuai. Hal ini menyederhanakan konsumsi downstream. Nilai yang valid: none (default): Tidak menghasilkan changelog tambahan. Konsumen downstream masih dapat membaca tabel Paimon dalam mode streaming, tetapi changelog-nya tidak lengkap (hanya berisi catatan `update_after` tanpa catatan `update_before` yang sesuai). input: Menulis aliran input ke file changelog, yang berfungsi sebagai changelog lengkap. full-compaction: Menghasilkan changelog lengkap selama setiap full compaction. lookup: Menghasilkan changelog lengkap sebelum setiap snapshot dikomit. Untuk informasi lebih lanjut tentang cara memilih produsen changelog, lihat Produksi Changelog.
full-compaction.delta-commits	Jumlah maksimum commit antara dua full compaction berturut-turut.	Integer	Tidak	Tidak ada	Menentukan jumlah maksimum commit snapshot yang diizinkan sebelum memicu full compaction.
lookup.cache-max-memory-size	Ukuran cache memori untuk tabel dimensi Paimon.	String	Tidak	256 MB	Parameter ini mengontrol ukuran cache baik untuk lookup tabel dimensi maupun produsen changelog `lookup`.
merge-engine	Mekanisme penggabungan catatan dengan kunci primer yang sama.	String	Tidak	deduplicate	Nilai yang valid: deduplicate: Menyimpan hanya catatan terbaru. partial-update: Memperbarui catatan yang ada dengan nilai non-null dari catatan terbaru. Kolom lain tetap tidak berubah. aggregation: Melakukan pre-agregasi dengan menggunakan fungsi agregat yang ditentukan. Untuk analisis mendetail tentang mesin merge, lihat Mesin Merge.
partial-update.ignore-delete	Menentukan apakah akan mengabaikan pesan delete (-D).	Boolean	Tidak	false	Nilai yang valid: true: Mengabaikan pesan delete. false: Memproses pesan delete. Anda harus mengonfigurasi strategi untuk menangani penghapusan menggunakan parameter seperti sequence.field untuk mencegah potensi error `IllegalStateException` atau `IllegalArgumentException`. Catatan Di Realtime Compute for Apache Flink 8.0.6 dan versi sebelumnya, parameter ini hanya berlaku dalam skenario partial update di mana `merge-engine = 'partial-update'`. Di Realtime Compute for Apache Flink 8.0.7 dan versi yang lebih baru, parameter ini juga kompatibel dengan skenario non-partial-update dan memiliki fungsi yang sama dengan parameter `ignore-delete`. Kami merekomendasikan Anda menggunakan `ignore-delete` sebagai gantinya. Tentukan apakah akan mengaktifkan parameter ini berdasarkan kebutuhan bisnis Anda dan apakah pesan delete tersebut diharapkan. Jika pesan delete tidak sesuai dengan semantik yang dimaksudkan oleh job Anda, lebih baik job gagal segera.
ignore-delete	Menentukan apakah akan mengabaikan pesan delete (-D).	Boolean	Tidak	false	Nilai yang valid sama dengan nilai untuk partial-update.ignore-delete. Catatan Parameter ini hanya didukung di Realtime Compute for Apache Flink 8.0.7 dan versi yang lebih baru. Parameter ini memiliki fungsi yang sama dengan `partial-update.ignore-delete`. Kami merekomendasikan Anda menggunakan `ignore-delete` dan menghindari konfigurasi kedua parameter secara bersamaan.
partition.default-name	Nama partisi default.	String	Tidak	__DEFAULT_PARTITION__	Nama partisi yang digunakan ketika nilai kolom partisi adalah null atau string kosong.
partition.expiration-check-interval	Seberapa sering sistem memeriksa partisi yang kedaluwarsa.	String	Tidak	1h	Untuk detailnya, lihat Cara mengonfigurasi kedaluwarsa partisi otomatis.
partition.expiration-time	Durasi setelah partisi kedaluwarsa.	String	Tidak	Tidak ada	Partisi kedaluwarsa ketika usianya melebihi nilai ini. Secara default, partisi tidak pernah kedaluwarsa. Sistem menghitung usia partisi dari nilai partisinya. Untuk detailnya, lihat Cara mengonfigurasi kedaluwarsa partisi otomatis.
partition.timestamp-formatter	String format untuk mengonversi string waktu menjadi timestamp.	String	Tidak	Tidak ada	Menentukan format untuk mengekstraksi usia partisi dari nilai partisi. Untuk detailnya, lihat Cara mengonfigurasi kedaluwarsa partisi otomatis.
partition.timestamp-pattern	String format untuk mengonversi nilai partisi menjadi string waktu.	String	Tidak	Tidak ada	Menentukan pola untuk mengekstraksi string waktu dari nilai partisi. Untuk detailnya, lihat Cara mengonfigurasi kedaluwarsa partisi otomatis.
scan.bounded.watermark	Nilai watermark yang menandai akhir pemindaian. Tabel sumber berhenti menghasilkan data ketika watermark-nya melebihi nilai ini.	Long	Tidak	Tidak ada	Tidak berlaku
scan.mode	Menentukan posisi konsumsi untuk tabel sumber Paimon.	String	Tidak	default	Untuk detailnya, lihat Cara mengatur posisi konsumsi untuk tabel sumber Paimon.
scan.snapshot-id	Menentukan snapshot tempat tabel sumber Paimon mulai mengonsumsi.	Integer	Tidak	Tidak ada	Untuk detailnya, lihat Cara mengatur posisi konsumsi untuk tabel sumber Paimon.
scan.timestamp-millis	Menentukan titik waktu tempat tabel sumber Paimon mulai mengonsumsi.	Integer	Tidak	Tidak ada	Untuk detailnya, lihat Cara mengatur posisi konsumsi untuk tabel sumber Paimon.
snapshot.num-retained.max	Jumlah maksimum snapshot terbaru yang dipertahankan.	Integer	Tidak	2147483647	Kedaluwarsa snapshot dipicu jika salah satu kondisi ini atau kondisi `snapshot.time-retained` terpenuhi, dan kondisi `snapshot.num-retained.min` juga terpenuhi.
snapshot.num-retained.min	Jumlah minimum snapshot terbaru yang dipertahankan.	Integer	Tidak	10	Tidak berlaku
snapshot.time-retained	Periode retensi untuk snapshot.	String	Tidak	1h	Kedaluwarsa snapshot dipicu jika salah satu kondisi ini atau kondisi `snapshot.num-retained.max` terpenuhi, dan kondisi `snapshot.num-retained.min` juga terpenuhi.
write-mode	Mode penulisan untuk tabel Paimon.	String	Tidak	change-log	Nilai yang valid: change-log: Tabel Paimon mendukung operasi insert, delete, dan update berdasarkan kunci primer. append-only: Tabel Paimon hanya menerima operasi insert dan tidak mendukung kunci primer. Mode ini lebih efisien daripada mode change-log. Untuk informasi lebih lanjut tentang mode penulisan, lihat Mode Penulisan.
scan.infer-parallelism	Menentukan apakah akan secara otomatis menginferensi parallelisme untuk tabel sumber Paimon.	Boolean	Tidak	true	Nilai yang valid: true: Secara otomatis menginferensi parallelisme tabel sumber Paimon berdasarkan jumlah bucket. false: Menggunakan parallelisme default yang dikonfigurasi di Realtime Compute for Apache Flink. Jika mode expert diaktifkan, job menggunakan parallelisme yang Anda konfigurasi.
scan.parallelism	Parallelisme untuk tabel sumber Paimon.	Integer	Tidak	Tidak ada	Catatan Parameter ini diabaikan jika expert mode diatur ke expert mode pada tab Deployment Details > > Resource Configuration job.
sink.parallelism	Parallelisme untuk tabel sink Paimon.	Integer	Tidak	Tidak ada	Catatan Parameter ini diabaikan jika expert mode diatur ke expert mode pada tab Deployment Details > > Resource Configuration job.
sink.clustering.by-columns	Menentukan kolom clustering untuk menulis ke tabel sink Paimon.	String	Tidak	Tidak ada	Untuk tabel append-only Paimon (tabel tanpa kunci primer), parameter ini mengaktifkan penulisan terkluster dalam job batch. Proses ini meningkatkan performa kueri dengan mengelompokkan data berdasarkan kolom yang ditentukan. Pisahkan beberapa nama kolom dengan koma (,), contohnya `'col1,col2'`. Untuk informasi lebih lanjut tentang clustering, lihat Dokumentasi Resmi Apache Paimon.
sink.delete-strategy	Menentukan strategi validasi untuk memastikan sistem menangani pesan retraksi (-D/-U) dengan benar.	Enum	Tidak	NONE	Nilai yang valid dan perilaku yang diharapkan dari operator sink saat menangani pesan retraksi: NONE (default): Tidak melakukan validasi. IGNORE_DELETE: Operator sink harus mengabaikan pesan -U dan -D. Tidak terjadi retraksi. NON_PK_FIELD_TO_NULL: Operator sink harus mengabaikan pesan -U. Untuk pesan -D, operator menyimpan kunci primer dan mengatur semua bidang non-kunci-primer lainnya ke null. Ini terutama digunakan untuk partial update ketika beberapa sink menulis ke tabel yang sama. DELETE_ROW_ON_PK: Operator sink harus mengabaikan pesan -U tetapi menghapus baris yang sesuai dengan kunci primer untuk pesan -D. CHANGELOG_STANDARD: Operator sink harus menghapus baris yang sesuai dengan kunci primer baik untuk pesan -U maupun -D. Catatan Parameter ini hanya didukung di Realtime Compute for Apache Flink 8.0.8 dan versi yang lebih baru. Perilaku sink aktual untuk retraksi ditentukan oleh parameter lain, seperti `ignore-delete` dan `merge-engine`. Parameter ini hanya memvalidasi apakah perilaku aktual sesuai dengan strategi yang dipilih. Jika terjadi ketidaksesuaian, job gagal dengan pesan error yang menyarankan koreksi.

Catatan

Untuk informasi lebih lanjut tentang opsi konfigurasi, lihat Dokumentasi Resmi Apache Paimon.

Detail fitur

Freshness dan konsistensi data

Tabel sink Paimon menggunakan protokol two-phase commit untuk meng-commit data selama setiap checkpoint job Flink. Oleh karena itu, freshness data ditentukan oleh interval checkpoint job Flink. Setiap commit menghasilkan hingga dua snapshot.

Saat dua job Flink menulis ke tabel Paimon yang sama secara konkuren, jika job tersebut menulis ke bucket yang berbeda, mereka mencapai konsistensi serializable. Jika job tersebut menulis ke bucket yang sama, mereka hanya mencapai isolasi snapshot. Artinya, data tabel mungkin merupakan campuran hasil dari kedua job tersebut, tetapi tidak terjadi kehilangan data.

Merge engine

Saat tabel sink Paimon menerima beberapa catatan dengan kunci primer yang sama, tabel tersebut menggabungkannya menjadi satu catatan untuk menjaga keunikan. Anda dapat mengontrol perilaku ini dengan mengatur parameter merge-engine. Tabel berikut menjelaskan mesin merge yang tersedia.

Mesin merge	Deskripsi
Deduplicate	Mesin deduplikasi adalah default. Untuk beberapa catatan dengan kunci primer yang sama, tabel sink Paimon hanya menyimpan catatan terbaru dan membuang yang lainnya. Catatan Jika catatan terbaru adalah pesan delete, semua catatan dengan kunci primer tersebut dibuang.
Partial Update	Mesin partial update memungkinkan Anda membangun catatan lengkap dengan memperbaruinya secara inkremental menggunakan beberapa pesan. Saat catatan baru dengan kunci primer yang sama tiba, nilai non-null-nya menimpa bidang yang sesuai dalam catatan yang ada. Mesin ini mengabaikan bidang yang bernilai null dalam catatan baru dan mempertahankan nilai yang ada. Sebagai contoh, asumsikan tabel sink Paimon menerima tiga catatan berikut secara berurutan: <1, 23.0, 10, NULL> <1, NULL, NULL, 'This is a book'> <1, 25.2, NULL, NULL> Jika kolom pertama adalah kunci primer, catatan gabungan akhirnya adalah <1, 25.2, 10, 'This is a book'>. Catatan Untuk membaca hasil partial update dalam mode streaming, Anda harus mengatur parameter `changelog-producer` ke `lookup` atau `full-compaction`. Mesin partial update tidak dapat memproses pesan delete. Anda dapat mengatur parameter `partial-update.ignore-delete` ke `true` untuk mengabaikan pesan delete.
Aggregation	Dalam beberapa kasus penggunaan, Anda mungkin hanya memerlukan nilai agregat catatan. Mesin agregasi menggabungkan catatan yang memiliki kunci primer yang sama dengan menggunakan fungsi agregat yang Anda tentukan. Untuk setiap kolom non-kunci-primer, Anda harus menentukan fungsi agregat menggunakan opsi `fields.<field-name>.aggregate-function`. Jika tidak, kolom tersebut secara default menggunakan fungsi agregat `last_non_null_value`. Sebagai contoh, pertimbangkan definisi tabel Paimon berikut. `CREATE TABLE MyTable ( product_id BIGINT, price DOUBLE, sales BIGINT, PRIMARY KEY (product_id) NOT ENFORCED ) WITH ( 'merge-engine' = 'aggregation', 'fields.price.aggregate-function' = 'max', 'fields.sales.aggregate-function' = 'sum' );` Kolom `price` diagregasi menggunakan fungsi `max`, dan kolom `sales` diagregasi menggunakan fungsi `sum`. Diberikan dua catatan input, <1, 23.0, 15> dan <1, 30.2, 20>, hasil akhirnya adalah <1, 30.2, 35>. Fungsi agregat yang didukung dan tipe datanya adalah: sum: Mendukung DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, dan DOUBLE. min dan max: Mendukung DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DATE, TIME, TIMESTAMP, dan TIMESTAMP_LTZ. last_value dan last_non_null_value: Mendukung semua tipe data. listagg: Mendukung STRING. bool_and dan bool_or: Mendukung BOOLEAN. Catatan Hanya fungsi `sum` yang mendukung retraksi dan penghapusan; fungsi agregat lain tidak. Jika Anda ingin kolom tertentu mengabaikan pesan retraksi dan delete, Anda dapat mengatur `'fields.${field_name}.ignore-retract'='true'`. Untuk membaca hasil agregasi dalam mode streaming, Anda harus mengatur parameter `changelog-producer` ke `lookup` atau `full-compaction`.

Produsen changelog

Atur parameter changelog-producer untuk mengonfigurasi Paimon agar menghasilkan changelog lengkap (di mana setiap catatan update_after memiliki catatan update_before yang sesuai) untuk aliran input apa pun. Tabel berikut menjelaskan produsen changelog yang tersedia. Untuk detail lebih lanjut, lihat dokumentasi resmi Apache Paimon.

Produsen	Deskripsi
None	Saat Anda mengatur `changelog-producer` ke `none` (default), tabel sumber Paimon downstream hanya melihat status terbaru data untuk kunci primer tertentu. Changelog yang tidak lengkap ini menyulitkan konsumen untuk melakukan perhitungan yang benar karena mereka tidak dapat menentukan status data sebelumnya, hanya apakah data tersebut dihapus atau status terbarunya. Sebagai contoh, jika konsumen downstream perlu menghitung jumlah suatu kolom dan hanya melihat nilai terbaru 5, konsumen tersebut tidak dapat menentukan cara memperbarui total. Jika nilai sebelumnya adalah 4, jumlah harus bertambah 1; jika nilai sebelumnya adalah 6, jumlah harus berkurang 1. Konsumen yang sensitif terhadap catatan `update_before` sebaiknya tidak menggunakan produsen `none`, tetapi produsen changelog lain menimbulkan overhead performa. Catatan Jika konsumen downstream Anda, seperti database, tidak sensitif terhadap data `update_before`, Anda dapat menggunakan produsen `none`. Konfigurasikan produsen changelog berdasarkan kebutuhan spesifik Anda.
Input	Saat Anda mengatur `changelog-producer` ke `input`, tabel sink melakukan dual-write aliran input ke file changelog. Gunakan produsen ini hanya ketika aliran input itu sendiri sudah merupakan changelog lengkap, seperti data dari Change Data Capture (CDC).
Lookup	Saat Anda mengatur `changelog-producer` ke `lookup`, tabel sink menggunakan mekanisme kueri titik, mirip dengan lookup tabel dimensi, untuk menghasilkan changelog lengkap untuk snapshot saat ini sebelum dikomit. Produsen ini menghasilkan changelog lengkap dari aliran input apa pun. Dibandingkan dengan produsen `full-compaction`, produsen `lookup` menawarkan timeliness yang lebih baik untuk changelog tetapi mengonsumsi lebih banyak sumber daya secara keseluruhan. Gunakan opsi ini untuk kasus penggunaan yang memerlukan freshness data tinggi (misalnya, tingkat menit).
Full Compaction	Saat Anda mengatur `changelog-producer` ke `full-compaction`, tabel sink menghasilkan changelog lengkap selama setiap full compaction. Produsen ini menghasilkan changelog lengkap dari aliran input apa pun. Interval untuk full compaction ditentukan oleh parameter `full-compaction.delta-commits`. Dibandingkan dengan produsen `lookup`, produsen `full-compaction` memiliki latensi lebih tinggi tetapi memanfaatkan proses full compaction yang sudah ada tanpa menambahkan komputasi tambahan. Hal ini menghasilkan konsumsi sumber daya keseluruhan yang lebih rendah. Gunakan opsi ini untuk kasus penggunaan dengan persyaratan freshness data rendah (misalnya, per jam).

Mode penulisan

Tabel Paimon mendukung mode penulisan berikut.

Mode

Deskripsi

Change-log

Mode penulisan change-log adalah default untuk tabel Paimon. Mode ini mendukung operasi insert, delete, dan update berdasarkan kunci primer. Anda juga dapat menggunakan mesin merge dan produsen changelog dalam mode ini.

Append-only

Mode penulisan append-only hanya mendukung penyisipan data dan tidak menggunakan kunci primer. Mode ini lebih efisien daripada mode change-log dan dapat digunakan sebagai alternatif antrian pesan dalam skenario dengan persyaratan freshness data moderat (misalnya, freshness tingkat menit).

Untuk deskripsi mendetail tentang mode penulisan append-only, lihat dokumentasi resmi Apache Paimon. Saat menggunakan mode ini, perhatikan hal berikut:

Atur parameter bucket-key sesuai kebutuhan Anda. Jika tidak, tabel Paimon melakukan bucketing data berdasarkan nilai semua kolom, yang tidak efisien secara komputasi.
Mode penulisan append-only dapat menjamin urutan output data sampai batas tertentu. Urutan output spesifiknya adalah sebagai berikut:
1. Untuk catatan dari partisi berbeda: jika parameter scan.plan-sort-partition diatur, catatan dari partisi dengan nilai lebih kecil di-output terlebih dahulu. Jika tidak, catatan dari partisi yang dibuat lebih awal di-output terlebih dahulu.
2. Untuk catatan dari partisi dan bucket yang sama, catatan yang ditulis lebih awal di-output terlebih dahulu.
3. Untuk catatan dari partisi yang sama tetapi bucket berbeda, urutan output tidak dijamin karena bucket berbeda diproses oleh tugas konkuren yang berbeda.

Target untuk CTAS dan CDAS

Tabel Paimon mendukung sinkronisasi data real-time untuk tabel tunggal atau seluruh database. Perubahan skema di tabel upstream juga disinkronkan ke tabel Paimon secara real-time. Untuk detailnya, lihat Mengelola tabel Paimon dan Mengelola Paimon Catalog.

Pemangkasan baca Variant

Hanya membaca bidang Variant yang direferensikan oleh kueri, mengurangi overhead I/O dan memori.

Cara mengaktifkan

Parameter	Nilai default	Deskripsi
`variant.read.pushdown.enabled`	false	Atur ke `true` untuk mengaktifkan pemangkasan baca Variant.

Prasyarat

Data Variant harus ditulis oleh versi saat ini. Data Variant yang ditulis oleh versi sebelumnya atau mesin lain tidak mendukung pemangkasan baca.

Skenario yang didukung

Pemangkasan baca berlaku saat SQL mengakses bidang Variant dengan kunci string, misalnya:

SELECT v['a']            FROM t;   -- Satu level
SELECT v['a']['b']       FROM t;   -- Bersarang
SELECT v['a'], v['b']    FROM t;   -- Beberapa bidang
SELECT id, v['a']        FROM t;   -- Dicampur dengan kolom biasa
SELECT v['a'] + 1        FROM t;   -- Direferensikan dalam ekspresi

Skenario yang tidak didukung

Mereferensikan seluruh Variant secara langsung, seperti SELECT v FROM t.
Menggabungkan referensi langsung dengan akses bidang, seperti SELECT v, v['a'] FROM t.
Akses indeks array, seperti v[0].
Skema tabel berisi bidang Row bersarang. Dalam kasus ini, pemangkasan baca tidak berlaku untuk bidang Variant apa pun di tabel.

Gunakan Paimon sebagai tabel dimensi

Tabel Paimon dapat digunakan sebagai tabel dimensi. Untuk sintaks JOIN, lihat Pernyataan JOIN tabel dimensi.

Secara default, lookup memuat semua data di setiap instans paralel. Pendekatan ini hanya cocok untuk tabel dimensi kecil. Untuk tabel dimensi besar, gunakan solusi Shuffle Lookup yang dijelaskan di bawah.

Tabel dimensi terpartisi

Jika tabel dimensi Anda terpartisi dan Anda hanya memerlukan data dari satu atau dua partisi terbaru, Anda dapat menggunakan fitur pemuatan partisi dinamis:

SELECT * FROM T
JOIN DIM /*+ OPTIONS('lookup.dynamic-partition'='max_pt()', 'lookup.dynamic-partition.refresh-interval'='1 h') */
FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col = D.col;

Parameter	Tipe data	Nilai default	Deskripsi
lookup.dynamic-partition	String	T/A	`max_pt()`: hanya memuat partisi terbaru. `max_two_pt()`: hanya memuat dua partisi terbaru.
lookup.dynamic-partition.refresh-interval	Durasi	1 h	Interval sistem memeriksa pembaruan partisi di tabel dimensi.

Tabel dimensi besar: tabel fixed-bucket

Hanya didukung di VVR 8.0.8 dan versi yang lebih baru. Untuk tabel fixed-bucket (bucket > 0), Anda dapat menggunakan Shuffle Lookup untuk mendistribusikan data berdasarkan kunci bucket ke instans paralel, sehingga setiap instans hanya memuat data di bucket yang ditugaskan:

SELECT /*+ LOOKUP('table'='D', 'shuffle'='true') */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;

Catatan

Kunci join harus merupakan kunci bucket. Kunci bucket secara default adalah kunci primer.
Hanya tabel fixed-bucket (bucket > 0) yang mendukung fitur ini.

Tabel dimensi besar: tabel non-fixed-bucket

Hanya didukung di VVR 8.0.10 dan versi yang lebih baru. Untuk tabel dynamic-bucket atau tabel append, Anda dapat menggunakan SHUFFLE_HASH atau REPLICATED_SHUFFLE_HASH sehingga setiap instans paralel membaca semua data tetapi hanya menyimpan bagian yang dibutuhkan:

-- Shuffle Hash
SELECT /*+ SHUFFLE_HASH(D) */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;

-- Replicated Shuffle Hash
SELECT /*+ REPLICATED_SHUFFLE_HASH(D) */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;

Untuk informasi lebih lanjut tentang SHUFFLE_HASH dan REPLICATED_SHUFFLE_HASH, lihat Pernyataan JOIN tabel dimensi.

Ingesti Data

Anda dapat menggunakan konektor Paimon sebagai sink dalam job Ingesti Data YAML.

Sintaks

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: filesystem
  catalog.properties.warehouse: /path/warehouse

Parameter

Parameter	Deskripsi	Wajib	Tipe	Default	Catatan
type	Tipe konektor.	Ya	STRING	Tidak ada	Nilainya harus `paimon`.
name	Nama sink.	Tidak	STRING	Tidak ada
catalog.properties.metastore	Tipe katalog Paimon.	Tidak	STRING	filesystem	Nilai yang valid: filesystem (default) rest (hanya mendukung Data Lake Formation (DLF), bukan DLF-Legacy)
catalog.properties.*	Parameter untuk membuat katalog Paimon.	Tidak	STRING	Tidak ada	Untuk informasi lebih lanjut, lihat Mengelola Paimon Catalog.
table.properties.*	Parameter untuk membuat tabel Paimon.	Tidak	STRING	Tidak ada	Untuk informasi lebih lanjut, lihat Opsi tabel Paimon.
catalog.properties.warehouse	Direktori root untuk penyimpanan file.	Tidak	STRING	Tidak ada	Parameter ini hanya berlaku saat `catalog.properties.metastore` diatur ke `filesystem`.
commit.user-prefix	Awalan username untuk meng-commit file data.	Tidak	STRING	Tidak ada	Catatan Kami merekomendasikan menetapkan username berbeda untuk job berbeda. Hal ini memudahkan identifikasi job yang menyebabkan konflik commit.
partition.key	Kunci partisi untuk tabel terpartisi.	Tidak	STRING	Tidak ada	Tabel berbeda dipisahkan dengan `;`, bidang berbeda dipisahkan dengan `,`, dan tabel serta bidang dipisahkan dengan `:`. Contohnya: `testdb.table1:id1,id2;testdb.table2:name`.
sink.cross-partition-upsert.tables	Daftar tabel yang memerlukan cross-partition upsert, di mana kunci primer tidak mencakup semua kunci partisi.	Tidak	STRING	Tidak ada	Format: Gunakan titik koma `;` untuk memisahkan nama tabel. Rekomendasi performa: Operasi ini intensif sumber daya. Buat job terpisah untuk tabel-tabel ini. Penting Anda harus mencantumkan semua tabel yang memenuhi kriteria. Menghilangkan nama tabel akan mengakibatkan duplikasi data.
sink.commit.parallelism	Menentukan parallelisme operator Commit.	Tidak	INTEGER	Tidak ada	Jika operator Commit menjadi bottleneck, gunakan parameter ini untuk meningkatkan parallelism-nya dan meningkatkan performa. Parameter ini hanya didukung di Realtime Compute for Apache Flink 11.6 dan versi yang lebih baru. Catatan Menyetel parameter ini mengubah parallelisme operator. Saat me-restart job stateful, Anda harus menentukan `AllowNonRestoredState` agar job dapat mengabaikan sebagian state operator.

Gunakan katalog yang sudah ada

Mulai dari Realtime Compute for Apache Flink 11.5, Anda dapat langsung mereferensikan katalog Paimon bawaan dari halaman Data Management dalam job Ingesti Data Flink CDC. Hal ini mengurangi konfigurasi manual.

sink:
  type: paimon
  using.built-in-catalog: paimon_dlf_catalog
  catalog.properties.fs.oss.endpoint: oss-cn-beijing-internal.aliyuncs.com

Job Ingesti Data dapat secara otomatis menggunakan kembali semua parameter dalam katalog Paimon. Hal ini setara dengan mengonfigurasi secara manual parameter yang diawali dengan catalog.properties. dalam job YAML.

Untuk meng-override parameter yang digunakan kembali secara otomatis, atur secara eksplisit dalam job YAML. Konfigurasi YAML eksplisit memiliki prioritas lebih tinggi. Sebagai contoh, dalam contoh di atas, parameter fs.oss.endpoint menggunakan nilai dari job YAML, meng-override nilai di paimon_dlf_catalog.

Contoh

Saat menggunakan Paimon sebagai sink Ingesti Data, rujuk contoh berikut untuk mengonfigurasi job berdasarkan tipe katalog Paimon Anda.

Contoh konfigurasi untuk menulis ke Object Storage Service (OSS) dengan katalog Paimon filesystem:

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: paimon
  name: Paimon Sink
  catalog.properties.metastore: filesystem
  catalog.properties.warehouse: oss://default/test
  catalog.properties.fs.oss.endpoint: oss-cn-beijing-internal.aliyuncs.com
  catalog.properties.fs.oss.accessKeyId: xxxxxxxx
  catalog.properties.fs.oss.accessKeySecret: xxxxxxxx

Untuk informasi tentang parameter yang diawali dengan catalog.properties, lihat Membuat Katalog Filesystem Paimon.

Contoh konfigurasi untuk menulis ke Data Lake Formation (DLF) dengan katalog Paimon rest:

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: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Opsional) Aktifkan deletion vectors untuk meningkatkan performa baca.
  table.properties.deletion-vectors.enabled: true

Untuk informasi tentang parameter yang diawali dengan catalog.properties, lihat Parameter Konfigurasi Katalog Flink CDC.

Perubahan Skema

Saat digunakan sebagai sink Ingesti Data, Paimon mendukung event perubahan skema berikut:

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

Catatan

Jika tabel Paimon downstream sudah ada, job menulis ke skema yang ada dan tidak mencoba membuat tabel lagi.

Realtime Compute for Apache Flink:Konektor Paimon

Informasi latar belakang

Fitur utama

Batasan dan rekomendasi

SQL

Sintaks

Parameter WITH

Detail fitur

Freshness dan konsistensi data

Merge engine

Produsen changelog

Mode penulisan

Target untuk CTAS dan CDAS

Pemangkasan baca Variant

Gunakan Paimon sebagai tabel dimensi

Tabel dimensi terpartisi

Tabel dimensi besar: tabel fixed-bucket

Tabel dimensi besar: tabel non-fixed-bucket

Ingesti Data

Sintaks

Parameter

Gunakan katalog yang sudah ada

Contoh

Perubahan Skema

FAQ

Versi Apache Paimon	VVR
1.3.1	11.5, 11.6
1.3	11.4
1.2	11.2, 11.3
1.1	11.1
1.0	8.0.11
0.9	8.0.7, 8.0.8, 8.0.9, 8.0.10