Realtime Compute for Apache Flink:Versioned Merge Engine

Ikhtisar Versioned Merge Engine

• Merge policy

  Menggabungkan catatan berdasarkan nomor versi atau timestamp event. Untuk setiap kunci primary, engine ini menyimpan catatan dengan nomor versi tertinggi atau timestamp terbaru.

• Konfigurasi

  Untuk mengaktifkan engine ini, atur properti tabel 'table.merge-engine' = 'versioned' saat membuat tabel.

• Mekanisme pembaruan

  • Pembaruan hanya terjadi jika nilai bidang yang ditentukan—seperti nomor versi—pada catatan baru lebih besar dari atau sama dengan nilai yang sudah ada.

  • Jika nilai baru lebih kecil dari nilai yang sudah ada atau bernilai null, engine tidak memicu pembaruan.

• Skenario

  • Menghapus duplikat data dan menggabungkan data yang tidak berurutan. Engine ini merupakan alternatif efektif terhadap transformasi deduplicate dalam Pemrosesan aliran.

  • Menjamin konsistensi akhir dengan data masukan, artinya sistem secara otomatis menyelesaikan ketidaksesuaian data dalam jendela waktu tertentu untuk mencapai keadaan konsisten.

• Batasan

  • Operasi UPDATE dan DELETE tidak didukung.

  • Pembaruan parsial tidak didukung.

  • Event UPDATE_BEFORE dan DELETE dalam change log secara otomatis diabaikan.

Contoh

Kolom versi

Kolom versi menyimpan nomor versi atau timestamp event dari suatu catatan data. Kolom ini digunakan untuk memilih versi terbaru dari suatu catatan dan wajib disediakan agar Versioned Merge Engine dapat diaktifkan.

'table.merge-engine' = 'versioned',
'table.merge-engine.versioned.ver-column' = '<column_name>'

Kolom versi mendukung tipe data berikut:

• INT

• BIGINT

• TIMESTAMP

• TIMESTAMP(p) (dengan presisi)

• TIMESTAMP_LTZ (timestamp zona waktu lokal)

• TIMESTAMP_LTZ(p) (timestamp zona waktu lokal dengan presisi)

Contoh SQL

CREATE TABLE VERSIONED (
    a INT NOT NULL PRIMARY KEY NOT ENFORCED,
    b STRING, 
    ts BIGINT
 ) WITH (
    'table.merge-engine' = 'versioned',
    'table.merge-engine.versioned.ver-column' = 'ts'
);

INSERT INTO VERSIONED (a, b, ts) VALUES (1, 'v1', 1000);

-- Masukkan data dengan ts < 1000. Catatan tidak diperbarui.
INSERT INTO VERSIONED (a, b, ts) VALUES (1, 'v2', 999);
SELECT * FROM VERSIONED WHERE a = 1;
-- Output
-- +---+-----+------+
-- | a | b   | ts   |
-- +---+-----+------+
-- | 1 | v1  | 1000 |
-- +---+-----+------+


-- Masukkan data dengan ts > 1000. Catatan diperbarui.
INSERT INTO VERSIONED (a, b, ts) VALUES (1, 'v3', 2000);
SELECT * FROM VERSIONED WHERE a = 1;
-- Output
-- +---+-----+------+
-- | a | b   | ts   |
-- +---+-----+------+
-- | 1 | v3  | 2000 |
-- +---+-----+------+

-- Masukkan data dengan ts null. Catatan tidak diperbarui.
INSERT INTO VERSIONED (a, b, ts) VALUES (1, 'v4', CAST(null as BIGINT));
SELECT * FROM VERSIONED WHERE a = 1;
-- Output
-- +---+-----+------+
-- | a | b   | ts   |
-- +---+-----+------+
-- | 1 | v3  | 2000 |
-- +---+-----+------+