Realtime Compute for Apache Flink:Penyimpanan aliran Fluss

Ikhtisar

Tabel kunci primer Fluss mendukung konsumsi lengkap dan inkremental. Sistem pertama-tama membaca snapshot historis lengkap dari tabel tersebut. Setelah pembacaan snapshot selesai, sistem secara otomatis dan mulus beralih ke konsumsi aliran inkremental. Proses ini memastikan kelengkapan dan konsistensi data.

Mode konsumsi

Anda dapat mengontrol titik awal konsumsi menggunakan parameter scan.startup.mode.

Contoh SQL

-- Konsumsi lengkap dan inkremental (mode bawaan)
SELECT * FROM `fluss_catalog`.`fluss_db`.orders;

-- Mengonsumsi dari timestamp tertentu menggunakan petunjuk
SELECT * FROM `fluss_catalog`.`fluss_db`.orders
  /*+ OPTIONS('scan.startup.mode' = 'timestamp', 'scan.startup.timestamp' = '2025-01-01 00:00:00') */;

-- Mengonsumsi dari offset terbaru
SELECT * FROM `fluss_catalog`.`fluss_db`.orders
  /*+ OPTIONS('scan.startup.mode' = 'latest') */;

Langganan log

Tabel kunci primer Fluss secara native mendukung langganan log CDC (Change Data Capture). Hal ini memungkinkan pekerjaan Flink downstream mengonsumsi aliran changelog lengkap (event INSERT, UPDATE_BEFORE, UPDATE_AFTER, dan DELETE) secara real-time tanpa memerlukan konfigurasi binlog terpisah.

Artinya:

• Semua perubahan pada tabel kunci primer secara otomatis menghasilkan log CDC.

• Aplikasi downstream dapat langsung berlangganan ke tabel kunci primer untuk menerima aliran changelog lengkap guna sinkronisasi data real-time atau ETL streaming.

Contoh SQL

-- Baca tabel kunci primer secara langsung untuk mendapatkan aliran changelog CDC
SELECT * FROM `fluss_catalog`.`fluss_db`.orders;
-- Output mencakup event INSERT, UPDATE, dan DELETE lengkap.

Tabel virtual

Selain berlangganan langsung ke aliran CDC, Fluss menyediakan mekanisme tabel virtual. Anda dapat mengakses changelog dengan metadata dengan menambahkan akhiran tertentu ke nama tabel dasar, yang menghilangkan kebutuhan akan penyimpanan data tambahan. Sistem menghasilkan tabel virtual ini secara otomatis, sehingga tidak perlu dibuat secara manual.

-- Baca tabel virtual $changelog untuk tabel orders
SELECT * FROM `fluss_catalog`.`fluss_db`.orders$changelog;

-- Mengonsumsi dari offset terbaru menggunakan petunjuk
SELECT * FROM `fluss_catalog`.`fluss_db`.orders$changelog
  /*+ OPTIONS('scan.startup.mode' = 'latest') */;

-- Filter untuk event penghapusan
SELECT * FROM `fluss_catalog`.`fluss_db`.orders$changelog
WHERE _change_type = 'delete';

-- Baca tabel virtual $binlog untuk tabel users
SELECT * FROM `fluss_catalog`.`fluss_db`.users$binlog;

-- Dapatkan nilai before dan after suatu bidang dari event update
SELECT
  `before`.user_name AS old_name,
  `after`.user_name  AS new_name,
  _commit_timestamp  AS change_time
FROM `fluss_catalog`.`fluss_db`.users$binlog
WHERE _change_type = 'update';

Ikhtisar

Fluss mendukung pembaruan parsial, yang memungkinkan beberapa aliran data menulis secara independen ke kolom masing-masing dalam tabel kunci primer lebar yang sama. Mesin Fluss secara otomatis menggabungkan bidang-bidang tersebut pada level catatan berdasarkan kunci primer. Pendekatan ini menghindari kebutuhan untuk memelihara state join multi-aliran yang kompleks di Flink dan mencegah masalah terkait ukuran state yang besar.

Kasus penggunaan

Kasus penggunaan khas adalah membangun tabel lebar profil pengguna real-time. Berbagai sumber data, seperti sistem rekomendasi, log impresi, log klik, dan data keranjang belanja, dapat memperbarui bidang masing-masing dalam tabel profil pengguna. Proses ini menciptakan tampilan 360 derajat lengkap tentang pengguna.

Batasan

• Tabel target harus berupa tabel kunci primer.

• Operasi tulis setiap aliran data harus mencakup semua bidang kunci primer.

• Topologi komputasi tidak boleh mengandung node join multi-aliran. Fluss menangani operasi merge langsung di lapisan penyimpanan.

Contoh SQL

-- 1. Buat tabel lebar
CREATE TABLE IF NOT EXISTS `fluss_catalog`.fluss_db.user_rec_wide (
    user_id   STRING,
    item_id   STRING,
    rec_score DOUBLE,
    imp_cnt   INT,
    click_cnt INT,
    PRIMARY KEY (user_id, item_id) NOT ENFORCED
) WITH ('bucket.num' = '3');

-- 2. Gunakan beberapa aliran untuk menulis ke bidang masing-masing
-- Aliran 1: Menulis skor rekomendasi
INSERT INTO `fluss_catalog`.fluss_db.user_rec_wide (user_id, item_id, rec_score)
SELECT user_id, item_id, rec_score 
FROM `fluss_catalog`.fluss_db.recommendations;

-- Aliran 2: Menulis jumlah impresi
INSERT INTO `fluss_catalog`.fluss_db.user_rec_wide (user_id, item_id, imp_cnt)
SELECT user_id, item_id, imp_cnt 
FROM `fluss_catalog`.fluss_db.impressions;

-- Aliran 3: Menulis jumlah klik
INSERT INTO `fluss_catalog`.fluss_db.user_rec_wide (user_id, item_id, click_cnt)
SELECT user_id, item_id, click_cnt 
FROM `fluss_catalog`.fluss_db.clicks;

Tiga pernyataan INSERT di atas dapat dijalankan sebagai pekerjaan Flink terpisah. Lapisan penyimpanan Fluss secara otomatis menggabungkan bidang-bidang tersebut untuk kunci primer yang sama menjadi baris-baris lengkap.

Ikhtisar

Delta Join adalah paradigma join aliran-aliran generasi berikutnya yang disediakan oleh Fluss. Paradigma ini memindahkan operasi pembaruan data ke lapisan penyimpanan Fluss, memanfaatkan tabel kunci primer dan indeks sekunder untuk lookup yang efisien. Desain ini menghilangkan kebutuhan untuk memelihara state join di Flink.

Keunggulan utama:

• Tidak ada state join: Menghilangkan penyimpanan state berlebih dan menghindari masalah garbage collection (GC) yang disebabkan oleh ukuran state besar.

• Konsumsi resource rendah: Mengurangi konsumsi memori dan CPU Flink hingga lebih dari 86% dalam pengujian.

• Stabilitas dan efisiensi lebih baik: Menghindari bottleneck kinerja yang disebabkan oleh pembengkakan state.

Perbandingan dengan join tradisional

Batasan

• Baik tabel kiri maupun kanan harus berupa tabel kunci primer Fluss (tabel partisi didukung).

• Kunci bucket harus merupakan awalan dari kunci primer.

• Kunci join harus mencakup kunci bucket. Untuk tabel partisi, kunci join juga harus mencakup kunci partisi.

Contoh SQL

1. Buat tabel

CREATE TABLE orders (
  user_id BIGINT,
  order_date DATE,
  order_id BIGINT,
  amount DECIMAL(10, 2),
  PRIMARY KEY (user_id, order_id, order_date) NOT ENFORCED
) WITH ('bucket.key' = 'user_id, order_id');

CREATE TABLE order_enhance (
  user_id BIGINT,
  order_id BIGINT,
  risk_level TINYINT,
  is_fraud BOOLEAN,
  update_time TIMESTAMP(3),
  PRIMARY KEY (user_id, order_id) NOT ENFORCED
) WITH ('bucket.key' = 'user_id, order_id');

2. Jalankan Delta Join

SELECT * FROM orders o
INNER JOIN order_enhance e
ON o.user_id = e.user_id AND o.order_id = e.order_id;

3. Optimalkan kinerja lookup dengan petunjuk

SELECT * FROM orders o /*+ OPTIONS('client.lookup.queue-size'='2560') */
INNER JOIN order_enhance e /*+ OPTIONS('client.lookup.queue-size'='2560') */
ON o.user_id = e.user_id AND o.order_id = e.order_id;

Konfigurasi parameter pekerjaan

Untuk menggunakan Delta Join, tambahkan konfigurasi berikut ke parameter runtime pekerjaan Flink Anda:

table.exec.delta-join.cache-enabled: 'true'
table.exec.async-lookup.buffer-capacity: '3000'
table.optimizer.delta-join.strategy: FORCE
table.exec.delta-join.left.cache-size: '100'
table.exec.delta-join.right.cache-size: '1000'

Setelah Anda mengaktifkan Delta Join, node Delta Join muncul dalam ikhtisar pekerjaan Flink, menunjukkan bahwa optimasi aktif.

Ikhtisar

Anda dapat menggunakan tabel kunci primer Fluss sebagai tabel dimensi Flink SQL. Lookup join memungkinkan Anda memperkaya data streaming dengan data dimensi secara real-time. Fluss mendukung lookup point KV dan pencarian indeks sekunder yang efisien, sehingga cocok untuk skenario join tabel dimensi berlatensi rendah.

Batasan

• Hanya tabel kunci primer yang dapat digunakan sebagai tabel dimensi.

• Kondisi join harus mencakup semua bidang kunci primer dari tabel dimensi.

Mode sinkron dan asinkron

Contoh SQL

-- Konfigurasikan cache lookup menggunakan petunjuk untuk melakukan join tabel dimensi
SELECT
  o.order_id,
  o.amount,
  u.user_name,
  p.product_name
FROM `fluss_catalog`.`fluss_db`.orders AS o
LEFT JOIN `fluss_catalog`.`fluss_db`.user_dim
  /*+ OPTIONS('lookup.cache' = 'PARTIAL', 'lookup.partial-cache.expire-after-write' = '60s', 'lookup.partial-cache.max-rows' = '10000') */
  FOR SYSTEM_TIME AS OF o.proctime AS u
  ON o.user_id = u.user_id
LEFT JOIN `fluss_catalog`.`fluss_db`.product_dim
  FOR SYSTEM_TIME AS OF o.proctime AS p
  ON o.product_id = p.product_id;

-- Gunakan mode lookup sinkron (cocok untuk debugging)
SELECT o.order_id, o.amount, d.user_name
FROM `fluss_catalog`.`fluss_db`.orders AS o
LEFT JOIN `fluss_catalog`.`fluss_db`.user_dim
  /*+ OPTIONS('lookup.async' = 'false') */
  FOR SYSTEM_TIME AS OF o.proctime AS d
  ON o.user_id = d.user_id;

Ikhtisar

Tabel kunci primer Fluss memungkinkan Anda mengonfigurasi mesin merge untuk mengontrol bagaimana sistem menggabungkan catatan dengan kunci primer yang sama. Gunakan parameter table.merge-engine untuk memilih salah satu strategi berikut:

Contoh SQL

-- Gunakan first_row untuk deduplikasi
CREATE TABLE IF NOT EXISTS `fluss_catalog`.fluss_db.event_dedup (
    event_id STRING,
    event_type STRING,
    event_time TIMESTAMP(3),
    payload STRING,
    PRIMARY KEY (event_id) NOT ENFORCED
) WITH (
  'table.merge-engine' = 'first_row'
);

-- Gunakan versioned untuk memperbarui berdasarkan kolom versi
CREATE TABLE IF NOT EXISTS `fluss_catalog`.fluss_db.user_profile (
    user_id BIGINT,
    user_name STRING,
    age INT,
    update_version BIGINT,
    PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'table.merge-engine' = 'versioned',
  'table.merge-engine.versioned.ver-column' = 'update_version'
);

Ikhtisar

Fluss mendukung tabel partisi. Anda dapat menggunakan PARTITIONED BY untuk membagi data ke partisi fisik yang berbeda berdasarkan kunci partisi. Ini ideal untuk mengelola siklus hidup data dan meningkatkan kinerja kueri berdasarkan waktu atau dimensi bisnis. Tabel partisi mendukung partisi single-field dan partisi multi-field, serta menawarkan tiga strategi manajemen partisi ortogonal: manajemen manual, manajemen otomatis, dan pembuatan dinamis.

Aturan dasar

• Tipe data kunci partisi harus berupa STRING.

• Untuk tabel kunci primer, kunci partisi harus merupakan subset dari kunci primer.

• Anda hanya dapat mengonfigurasi aturan partisi otomatis saat membuat tabel; Anda tidak dapat mengubahnya nanti.

Strategi manajemen partisi

Ketiga strategi ini ortogonal dan dapat digunakan secara kombinasi. Misalnya, Anda dapat mengaktifkan manajemen otomatis untuk pembuatan dan pembersihan berbasis waktu, serta pembuatan dinamis sebagai cadangan untuk menangani partisi yang tidak dibuat sebelumnya.

Contoh SQL

1. Buat tabel partisi single-field dengan manajemen partisi otomatis

CREATE TABLE `fluss_catalog`.fluss_db.site_access (
  site_id BIGINT,
  user_name STRING,
  pv BIGINT,
  event_day STRING,
  PRIMARY KEY (site_id, event_day) NOT ENFORCED
) PARTITIONED BY (event_day) WITH (
  'table.auto-partition.enabled' = 'true',
  'table.auto-partition.time-unit' = 'DAY',
  'table.auto-partition.num-precreate' = '5',
  'table.auto-partition.num-retention' = '30',
  'table.auto-partition.time-zone' = 'Asia/Shanghai'
);

2. Buat tabel partisi multi-field

CREATE TABLE `fluss_catalog`.fluss_db.region_access (
  access_id BIGINT,
  user_name STRING,
  event_date STRING,
  region STRING,
  PRIMARY KEY (access_id, event_date, region) NOT ENFORCED
) PARTITIONED BY (event_date, region) WITH (
  'table.auto-partition.enabled' = 'true',
  'table.auto-partition.key' = 'event_date',
  'table.auto-partition.time-unit' = 'DAY',
  'table.auto-partition.num-retention' = '7'
);

Catatan: Untuk partisi multi-field, Anda harus menentukan kunci berbasis waktu untuk partisi otomatis menggunakan table.auto-partition.key. Partisi multi-field hanya mendukung kedaluwarsa dan pembersihan otomatis, bukan pembuatan pra-otomatis. Parameter num-precreate dipaksakan menjadi 0.

3. Kelola partisi secara manual

-- Lihat semua partisi
SHOW PARTITIONS `fluss_catalog`.fluss_db.site_access;

-- Tambahkan partisi secara manual
ALTER TABLE `fluss_catalog`.fluss_db.site_access
  ADD PARTITION (event_day = '20250315');

-- Hapus partisi secara manual
ALTER TABLE `fluss_catalog`.fluss_db.site_access
  DROP PARTITION (event_day = '20250101');

4. Buat partisi secara dinamis

Saat Anda menulis data, jika partisi target tidak ada, klien secara otomatis membuatnya. Fitur ini diaktifkan secara bawaan.

-- Tulis data. Jika partisi '20250701' tidak ada, partisi tersebut dibuat secara otomatis.
INSERT INTO `fluss_catalog`.fluss_db.site_access
VALUES (1, 'hello', 100, '20250701');

Parameter klien client.writer.dynamic-create-partition.enabled (bawaan: true) mengontrol pembuatan dinamis. Fitur ini tunduk pada batasan tingkat kluster max.partition.num (bawaan: 1000) dan max.bucket.num (bawaan: 128.000).

Pemangkasan partisi

Saat Anda mengkueri tabel partisi dalam mode streaming atau batch, Fluss mendukung pemangkasan partisi. Dengan memfilter berdasarkan kunci partisi dalam klausa WHERE, kueri hanya membaca partisi yang cocok, sehingga mengurangi overhead I/O yang tidak perlu.

-- Baca hanya data dari partisi untuk tanggal tertentu
SELECT * FROM `fluss_catalog`.`fluss_db`.site_access
WHERE event_day = '20250315';

-- Filter tabel partisi multi-field berdasarkan region
SELECT * FROM `fluss_catalog`.`fluss_db`.region_access
WHERE region = 'US';

Parameter partisi otomatis

Praktik terbaik

• Pilih granularitas partisi yang tepat: Pilih granularitas waktu berdasarkan volume data dan pola kueri Anda. Untuk volume data besar dan rentang waktu kueri sempit, kami merekomendasikan partisi berdasarkan DAY atau HOUR. Untuk volume data lebih kecil, Anda dapat mempartisi berdasarkan MONTH atau QUARTER.

• Tetapkan kebijakan retensi yang wajar: Gunakan num-retention untuk mengontrol jumlah partisi historis. Hal ini membantu menghindari pembengkakan metadata dan mengurangi beban manajemen kluster akibat jumlah partisi yang berlebihan.

• Kombinasikan manajemen otomatis dan pembuatan dinamis: Gunakan manajemen otomatis untuk pembuatan dan pembersihan partisi rutin berbasis waktu, serta gunakan pembuatan dinamis sebagai cadangan untuk menangani data abnormal atau tertunda.

• Perhatikan batasan kluster: Parameter max.partition.num (bawaan: 1000) membatasi jumlah partisi per tabel. Jumlah total bucket dalam kluster dibatasi oleh max.bucket.num (bawaan: 128.000). Jumlah total bucket dalam tabel (jumlah partisi × bucket.num) tidak boleh melebihi batas ini.

Fluss menggunakan format penyimpanan kolom Apache Arrow untuk transmisi log. Hal ini mendukung mendorong operasi filter dan proyeksi ke lapisan penyimpanan, sehingga mengurangi volume transfer data dan dapat meningkatkan kinerja kueri hingga 10x.

Fitur ini sangat efektif dalam skenario berikut:

• Membaca hanya subset kolom dari tabel lebar (projection pushdown).

• Memfilter sejumlah besar data berdasarkan kondisi tertentu (predicate pushdown).

Format penyimpanan log dikontrol oleh parameter table.log.format, yang bawaannya adalah ARROW.

Ikhtisar

Fluss mendukung arsitektur stream-lake terpadu yang secara otomatis menyinkronkan data real-time dari Fluss ke data lake, seperti Apache Paimon. Hal ini mencapai penyimpanan bertingkat otomatis untuk data panas dan dingin, menyeimbangkan latensi rendah kueri real-time dengan biaya penyimpanan historis yang rendah.

Fitur utama:

• Secara otomatis menulis data streaming dari Fluss ke penyimpanan lake Paimon.

• Secara otomatis mengklasifikasikan data panas dan dingin. Data panas dipertahankan di penyimpanan aliran Fluss, sedangkan data dingin dipindahkan ke Paimon.

• Menyediakan antarmuka kueri Flink SQL terpadu untuk mengakses data di penyimpanan aliran dan penyimpanan lake secara bersamaan.

Parameter konfigurasi

Ikhtisar

Fluss mendukung eksplorasi data cepat pada tabel dalam mode batch Flink SQL, termasuk kueri LIMIT, lookup point kunci primer, dan agregasi COUNT(*). Anda dapat dengan cepat melihat data dalam tabel tanpa memulai pekerjaan streaming yang berjalan terus-menerus. Hal ini berguna untuk pengembangan, debugging, validasi data, dan troubleshooting.

Kueri LIMIT

Anda dapat menjalankan kueri LIMIT pada tabel kunci primer dan tabel append-only untuk dengan cepat mengambil sampel kecil data.

-- Lihat 10 catatan pertama dari tabel kunci primer
SELECT * FROM `fluss_catalog`.`fluss_db`.orders LIMIT 10;

-- Lihat 5 catatan pertama dari tabel append-only
SELECT * FROM `fluss_catalog`.`fluss_db`.events LIMIT 5;

-- Gabungkan dengan proyeksi untuk melihat hanya kolom tertentu
SELECT order_id, amount, order_time 
FROM `fluss_catalog`.`fluss_db`.orders LIMIT 20;

Lookup point kunci primer

Untuk tabel kunci primer, Anda dapat melakukan lookup point presisi dengan menentukan kunci primer dalam klausa WHERE. Hal ini memanfaatkan mesin penyimpanan KV Fluss untuk mencapai waktu respons tingkat milidetik.

-- Lakukan lookup point untuk satu catatan berdasarkan kunci primernya
SELECT * FROM `fluss_catalog`.`fluss_db`.orders
WHERE order_id = 1001;

-- Lakukan lookup point dengan kunci primer komposit
SELECT * FROM `fluss_catalog`.`fluss_db`.user_orders
WHERE user_id = 12345 AND order_id = 6789;

Agregasi COUNT(*)

Anda dapat menjalankan kueri COUNT(*) pada tabel kunci primer untuk dengan cepat mendapatkan jumlah total catatan.

-- Hitung jumlah total baris dalam tabel kunci primer
SELECT COUNT(*) FROM `fluss_catalog`.`fluss_db`.orders;