Pembaruan parsial (membangun tabel lebar) - Realtime Compute for Apache Flink

Topik ini menjelaskan cara menggunakan fitur pembaruan parsial.

Mengapa menggunakan pembaruan parsial untuk membangun tabel lebar

Dalam pipa data aliran tradisional, pembuatan tabel lebar sering kali memerlukan penggabungan beberapa aliran atau tabel berdasarkan kunci primary. Misalnya, sistem rekomendasi real-time perlu mengonsolidasikan pembaruan dari berbagai sumber—seperti preferensi pengguna, pembelian, klik, dan keranjang belanja—untuk membentuk tampilan pelanggan 360 derajat.

Meskipun Apache Flink mendukung join multi-aliran, pendekatan ini memiliki sejumlah tantangan dalam skenario berskala besar:

Ukuran state besar: Semua event harus di-cache hingga proses join selesai, yang dapat menyebabkan ukuran state terus bertambah tanpa batas.
Hambatan performa: Overhead checkpoint yang besar dan tekanan balik (backpressure) yang sering dapat mengurangi throughput dan stabilitas secara keseluruhan.
Debugging sulit: State yang kompleks dan tidak transparan membuat troubleshooting memakan waktu serta biaya tinggi.
Konsistensi data buruk: Penetapan Time-to-Live (TTL) dapat menyebabkan event dibuang terlalu dini, sehingga menghasilkan output yang salah.

Tantangan-tantangan tersebut menunjukkan perlunya metode baru untuk membangun tabel lebar yang menghindari join kompleks serta secara signifikan mengurangi konsumsi resource dan overhead maintenance.

Membangun tabel lebar dengan pembaruan parsial Fluss

Fluss memperkenalkan metode pembangunan tabel lebar berbasis mekanisme pembaruan parsial. Pendekatan ini memungkinkan Anda memperkaya data dengan informasi dimensi secara langsung dalam pekerjaan pemrosesan aliran, sehingga menghilangkan kebutuhan akan join multi-aliran yang kompleks.

Berbeda dengan metode tradisional, Fluss memungkinkan setiap sumber data memperbarui kolomnya sendiri secara independen dalam satu tabel lebar berdasarkan kunci primary. Misalnya, Anda dapat mendefinisikan tabel user_profile dengan user_id sebagai kunci primary, yang mencakup semua kolom yang mungkin ada. Setiap aliran data hanya perlu menulis data parsial yang dimilikinya, dan mesin penyimpanan Fluss secara otomatis menggabungkan pembaruan tersebut berdasarkan kunci primary.

Mekanisme dasarnya bekerja sebagai berikut: ketika pembaruan parsial tiba, Fluss menemukan catatan yang sudah ada untuk kunci primary tersebut dan hanya memperbarui kolom yang disediakan dalam event saat ini, sementara kolom lainnya tetap tidak berubah. Versi baru yang telah digabung ditulis ke disk secara real time, memastikan bahwa setiap catatan selalu dalam kondisi terbaru dan lengkap. Proses ini mirip dengan pembaruan inkremental dalam database, di mana hanya bidang yang berubah yang dimodifikasi, bukan seluruh catatan. Pendekatan ini secara drastis mengurangi beban state dan kompleksitas komputasi pada pekerjaan Flink, sehingga membuat pipa data lebih ringan, efisien, dan mudah dirawat.

Contoh

Prasyarat

Langkah 1: Buat tabel

Catalog Fluss bernama fluss-demo telah dibuat.

Login ke Konsol Realtime Compute for Apache Flink.
Pada kolom Actions ruang kerja target, klik Console.
Di panel navigasi sebelah kiri, pilih Development > Scripts > New Temporary Script.

Di database fluss default, buat tiga tabel untuk merepresentasikan sumber data berbeda dalam membangun tabel lebar, serta satu tabel lebar sebagai tabel hasil.

-- Recommendations – model scores
CREATE TABLE IF NOT EXISTS `fluss-demo`.fluss.recommendations (
    user_id  STRING,
    item_id  STRING,
    rec_score DOUBLE,
    rec_ts   TIMESTAMP(3),
    PRIMARY KEY (user_id, item_id) NOT ENFORCED
) WITH ('bucket.num' = '3');
-- Impressions – how often we showed something
CREATE TABLE IF NOT EXISTS `fluss-demo`.fluss.impressions (
    user_id STRING,
    item_id STRING,
    imp_cnt INT,
    imp_ts  TIMESTAMP(3),
    PRIMARY KEY (user_id, item_id) NOT ENFORCED
) WITH ('bucket.num' = '3');
-- Clicks – user engagement
CREATE TABLE IF NOT EXISTS `fluss-demo`.fluss.clicks (
    user_id  STRING,
    item_id  STRING,
    click_cnt INT,
    clk_ts    TIMESTAMP(3),
    PRIMARY KEY (user_id, item_id) NOT ENFORCED
) WITH ('bucket.num' = '3');
-- Result wide table
CREATE TABLE IF NOT EXISTS `fluss-demo`.fluss.user_rec_wide (
    user_id   STRING,
    item_id   STRING,
    rec_score DOUBLE,   -- updated by recs stream
    imp_cnt   INT,      -- updated by impressions stream
    click_cnt INT,      -- updated by clicks stream
    PRIMARY KEY (user_id, item_id) NOT ENFORCED
) WITH ('bucket.num' = '3');

Klik Run untuk membuat tabel.

Langkah 2: Buat pekerjaan aliran

Di panel navigasi sebelah kiri, pilih Development > ETL > New > New Blank Stream Draft.

Masukkan data dari tabel recommendations, impressions, dan clicks, yang menggunakan user_id sebagai kunci primary, ke dalam tabel lebar besar user_rec_wide yang juga menggunakan user_id sebagai kunci primary.

BEGIN STATEMENT SET
;
INSERT INTO `fluss-demo`.fluss.user_rec_wide (user_id, item_id, rec_score)
SELECT
    user_id
    ,item_id
    ,rec_score
FROM `fluss-demo`.fluss.recommendations
;
INSERT INTO `fluss-demo`.fluss.user_rec_wide (user_id, item_id, imp_cnt)
SELECT
    user_id
    ,item_id
    ,imp_cnt
FROM `fluss-demo`.fluss.impressions
;
-- Apply click counts
INSERT INTO `fluss-demo`.fluss.user_rec_wide (user_id, item_id, click_cnt)
SELECT
    user_id
    ,item_id
    ,click_cnt
FROM `fluss-demo`.fluss.clicks
;
END
;

Di pojok kanan atas, klik Deploy > Operations. Klik Start untuk menjalankan pekerjaan.

Setelah pekerjaan dimulai, halaman Status akan muncul. Health Score bernilai 100 dan Historical Restart Count bernilai 0. Topologi DAG menunjukkan bahwa tabel sumber adalah fluss-demo.fluss.recommendations, fluss-demo.fluss.impressions, dan fluss-demo.fluss.clicks, sedangkan tabel lebar target adalah user_rec_wide.

Perhatikan bahwa topologi komputasi tidak mengandung node join multi-aliran. Setiap tabel sumber menulis secara independen ke tabel lebar.

Langkah 3: Masukkan data sumber

Di halaman Scripts, buat skrip kueri kosong lainnya dan jalankan pernyataan SQL berikut untuk memasukkan data uji ke dalam tiga tabel sumber:

-- Recommendations – model scores
INSERT INTO `fluss-demo`.fluss.recommendations VALUES
    ('user_101','prod_501',0.92 , TIMESTAMP '2025-05-16 09:15:02'),
    ('user_101','prod_502',0.78 , TIMESTAMP '2025-05-16 09:15:05'),
    ('user_102','prod_503',0.83 , TIMESTAMP '2025-05-16 09:16:00'),
    ('user_103','prod_501',0.67 , TIMESTAMP '2025-05-16 09:16:20'),
    ('user_104','prod_504',0.88 , TIMESTAMP '2025-05-16 09:16:45');
-- Impressions – how often each (user,item) was shown
INSERT INTO `fluss-demo`.fluss.impressions VALUES
    ('user_101','prod_501', 3, TIMESTAMP '2025-05-16 09:17:10'),
    ('user_101','prod_502', 1, TIMESTAMP '2025-05-16 09:17:15'),
    ('user_102','prod_503', 7, TIMESTAMP '2025-05-16 09:18:22'),
    ('user_103','prod_501', 4, TIMESTAMP '2025-05-16 09:18:30'),
    ('user_104','prod_504', 2, TIMESTAMP '2025-05-16 09:18:55');
-- Clicks – user engagement
INSERT INTO `fluss-demo`.fluss.clicks VALUES
    ('user_101','prod_501', 1, TIMESTAMP '2025-05-16 09:19:00'),
    ('user_101','prod_502', 2, TIMESTAMP '2025-05-16 09:19:07'),
    ('user_102','prod_503', 1, TIMESTAMP '2025-05-16 09:19:12'),
    ('user_103','prod_501', 1, TIMESTAMP '2025-05-16 09:19:20'),
    ('user_104','prod_504', 1, TIMESTAMP '2025-05-16 09:19:25');

Langkah 4: Verifikasi hasil

Pilih Development > ETL > New > New Blank Stream Draft.
```
SELECT * FROM `fluss-demo`.fluss.user_rec_wide;
```
Klik tombol Debug. Anda akan melihat bahwa setelah ketiga tabel detail ditulis secara terpisah, sistem mengagregasi hasil komputasi aliran dan menghasilkan data yang benar untuk tabel lebar besar dengan user_id sebagai kunci primary.

Kueri terhadap tabel lebar user_rec_wide mengembalikan lima catatan dengan kolom user_id, item_id, rec_score, imp_cnt, dan click_cnt, yang mengonfirmasi bahwa data dari ketiga tabel sumber telah digabung dengan benar.