Realtime Compute for Apache Flink：Menangani peristiwa changelog yang tidak berurutan

Informasi latar belakang

Log biner (binlog) dalam database relasional seperti MySQL mencatat semua operasi modifikasi dalam database, termasuk operasi INSERT, UPDATE, dan DELETE. Demikian pula, changelog dalam Flink SQL mencatat semua perubahan data untuk memfasilitasi pemrosesan data inkremental.

Dalam MySQL, Anda dapat menggunakan binlog untuk cadangan data, pemulihan, sinkronisasi, dan replikasi. Misalnya, Anda dapat membaca dan mengurai catatan operasi dalam binlog untuk melakukan sinkronisasi dan replikasi data inkremental. Change Data Capture (CDC) adalah teknologi umum yang digunakan untuk sinkronisasi data. Alat CDC dapat memantau perubahan data dalam database dan mengonversi perubahan tersebut menjadi aliran peristiwa untuk memfasilitasi pemrosesan real-time. Anda dapat menggunakan alat CDC untuk mengirim perubahan data dari database relasional ke gudang data atau sistem lain untuk analisis dan pelaporan real-time. Beberapa alat CDC populer meliputi Debezium dan Maxwell. Apache Flink menambahkan dukungan CDC untuk mengatasi FLINK-15331, memungkinkan integrasi data CDC dari sistem eksternal untuk menerapkan sinkronisasi dan analisis data real-time.

Pembuatan dan pemrosesan peristiwa changelog

Seperti disebutkan dalam bagian "Informasi Latar Belakang" dari topik ini, peristiwa changelog dapat dihasilkan oleh sumber eksternal seperti file binlog dan alat CDC. Peristiwa changelog juga dapat dihasilkan oleh operasi internal Flink SQL. Aliran changelog yang hanya berisi peristiwa INSERT disebut append stream atau non-update stream. Aliran changelog yang berisi jenis peristiwa lain, seperti peristiwa UPDATE, disebut update stream. Beberapa operasi dalam Flink, seperti agregasi grup dan deduplikasi, dapat menghasilkan peristiwa UPDATE. Dalam kebanyakan kasus, operator yang menghasilkan peristiwa UPDATE mempertahankan status dan disebut sebagai operator stateful. Harap diperhatikan bahwa tidak semua operator stateful mendukung aliran update sebagai input. Misalnya, operator over aggregation dan interval join tidak mendukung aliran update.

Untuk informasi tentang operasi kueri yang didukung dalam Realtime Compute for Apache Flink menggunakan Ververica Runtime (VVR) 6.0 atau lebih baru, operator runtime yang sesuai, serta dukungan untuk mengonsumsi dan menghasilkan aliran update, lihat Eksekusi Kueri.

Jenis peristiwa changelog

Apache Flink memperkenalkan mekanisme retraction sebagaimana diusulkan dalam FLINK-6047 untuk mengimplementasikan algoritma pembaruan inkremental untuk operator SQL streaming. Dalam mekanisme ini, peristiwa dikategorikan sebagai INSERT atau DELETE. Pengkategorian ini berlaku untuk sumber data yang hanya mendukung peristiwa INSERT. Kemudian, Apache Flink merefaktor sistem peristiwa changelog untuk mengatasi FLINK-16987. Sistem yang direfaktor menggunakan jenis peristiwa changelog berikut untuk memfasilitasi integrasi dengan ekosistem CDC:

/**
 * Jenis baris dalam sebuah Changelog.
 */
@PublicEvolving
public enum RowKind {

	/**
	 * Operasi penyisipan.
	 */
	INSERT,

	/**
	 * Konten sebelumnya dari baris yang diperbarui.
	 */
	UPDATE_BEFORE,

	/**
	 * Konten baru dari baris yang diperbarui.
	 */
	UPDATE_AFTER,

	/**
	 * Operasi penghapusan.
	 */
	DELETE
}

Seperti ditunjukkan dalam kode sebelumnya, UPDATE_BEFORE dan UPDATE_AFTER merupakan jenis peristiwa terpisah. Apache Flink tidak menggabungkan jenis peristiwa ini menjadi tipe peristiwa UPDATE komposit karena alasan berikut:

• Peristiwa jenis UPDATE_BEFORE dan UPDATE_AFTER memiliki struktur yang sama. Satu-satunya perbedaan adalah properti RowKind, yang membuat serialisasi lebih mudah. Sebaliknya, jika tipe peristiwa UPDATE komposit digunakan, peristiwa tersebut menjadi heterogen atau menyamakan peristiwa INSERT dan DELETE dengan peristiwa UPDATE. Misalnya, peristiwa INSERT hanya berisi peristiwa UPDATE_AFTER, sedangkan peristiwa DELETE hanya berisi peristiwa UPDATE_BEFORE.

• Operasi pengacakan data, seperti join dan agregasi, sering terjadi dalam lingkungan terdistribusi. Ini memerlukan peristiwa UPDATE komposit dibagi menjadi peristiwa DELETE dan INSERT terpisah dalam skenario tertentu, seperti ditunjukkan dalam contoh berikut.

Contoh

Dalam skenario sampel ini, peristiwa UPDATE komposit dibagi menjadi peristiwa DELETE dan INSERT. Kode SQL berikut juga digunakan untuk mendemonstrasikan masalah peristiwa changelog yang tidak berurutan dan solusi yang sesuai, yang dijelaskan dalam bagian selanjutnya dari topik ini.

-- Tabel sumber CDC: s1 & s2
CREATE TEMPORARY TABLE s1 (
  id BIGINT, 
  level BIGINT,
  PRIMARY KEY(id) NOT ENFORCED
)WITH (...);

CREATE TEMPORARY TABLE s2 (
  id BIGINT, 
  attr VARCHAR, 
  PRIMARY KEY(id) NOT ENFORCED
)WITH (...);

-- tabel sink: t1
CREATE TEMPORARY TABLE t1 (
  id BIGINT, 
  level BIGINT, 
  attr VARCHAR,
  PRIMARY KEY(id) NOT ENFORCED
)WITH (...);

-- gabungkan s1 dan s2 dan masukkan hasilnya ke t1 
INSERT INTO t1
SELECT 
  s1.*, s2.attr
FROM s1 JOIN s2
ON s1.level = s2.id;

Dalam aliran changelog tabel s1, jika rekaman (id=1, level=10) dimasukkan pada waktu t0 dan diperbarui menjadi (id=1, level=20) pada waktu t1, tiga peristiwa terpisah dihasilkan. Tabel berikut menggambarkan peristiwa-peristiwa ini.

Kunci utama tabel s1 adalah id. Namun, operasi join memerlukan data diacak berdasarkan kolom level, seperti yang ditentukan dalam klausa ON.

Jika operator Join memiliki paralelisme 2, tiga peristiwa sebelumnya mungkin dikirim ke dua tugas. Dalam hal ini, bahkan jika tipe peristiwa UPDATE komposit digunakan, peristiwa tersebut masih perlu dibagi selama pengacakan untuk memastikan pemrosesan paralel.

Penyebab

Bagian ini menggunakan contoh sebelumnya dan menambahkan kondisi berikut: Operator Join menerima +I (id=10, attr='a1') dan +I (id=20, attr='b1') peristiwa dari tabel s2, kemudian menerima tiga peristiwa changelog dari tabel s1. Dalam lingkungan terdistribusi, urutan kedatangan peristiwa changelog sebelumnya ke operator Sink hilir dapat bervariasi karena operasi join dilakukan secara paralel oleh dua tugas. Tabel berikut menjelaskan kemungkinan urutan peristiwa.

Urutan peristiwa dalam Kasus 1 sama dengan pemrosesan berurutan. Dalam Kasus 2 dan Kasus 3, peristiwa changelog tiba di luar urutan di operator hilir, yang dapat menyebabkan hasil yang salah dalam Flink SQL. Dalam contoh ini, kunci utama tabel sink adalah id. Jika penyimpanan eksternal melakukan operasi upsert dalam Kasus 2 dan Kasus 3, rekaman dengan id 1 secara salah dihapus dari penyimpanan eksternal. Namun, hasil yang diharapkan adalah bahwa rekaman (id=1, level=20, attr='b1') ada.

Gunakan SinkUpsertMaterializer sebagai solusi

Dalam contoh ini, operator Join menghasilkan aliran update karena output berisi peristiwa INSERT (+I) dan peristiwa UPDATE (-U dan +U). Jika peristiwa changelog yang tidak berurutan tidak ditangani dengan benar, hasil akhirnya mungkin salah.