Realtime Compute for Apache Flink:Tabel kunci primer

Konsep utama

Tabel kunci primer di Fluss menegakkan keunikan pada kunci primer yang ditentukan dan mendukung operasi INSERT, UPDATE, dan DELETE.

Sebagai contoh, pernyataan Flink SQL berikut membuat tabel kunci primer yang menetapkan shop_id, user_id, dan ds sebagai kunci primer serta mendistribusikan data ke dalam empat bucket.

CREATE TABLE pk_table (
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT,
  ds STRING,
  PRIMARY KEY (shop_id,user_id,ds) NOT ENFORCED
) WITH (
  'bucket.num' = '4'
);

• Jika Anda menulis beberapa catatan dengan kunci primer yang sama, sistem secara default menyimpan catatan terakhir berdasarkan waktu pemrosesan. Untuk strategi penulisan lainnya, lihat Merge engines.

• Untuk tabel partisi, kunci primer harus mencakup semua bidang kunci partisi. Misalnya, jika kunci partisi adalah ds, maka ds harus menjadi bagian dari kunci primer. Karena data dipartisi berdasarkan kunci partisi, Fluss menjaga keunikan dalam setiap partisi. Hal ini mencegah adanya catatan dengan kunci primer yang sama di partisi berbeda.

Bucketing data

Untuk tabel kunci primer, Fluss menetapkan catatan ke suatu bucket berdasarkan nilai hash dari kunci bucket-nya.

• Kunci bucket harus merupakan subset dari kunci primer, tidak termasuk kunci partisi apa pun.

  Sebagai contoh, jika kunci primer adalah (shop_id, user_id, ds) dan kunci partisi adalah ds, kunci bucket dapat berupa 'bucket.key' = 'user_id,shop_id', 'bucket.key' = 'user_id', atau 'bucket.key' = 'shop_id'.

• Jika Anda tidak menentukan kunci bucket, Fluss menggunakan kolom kunci primer (tidak termasuk kunci partisi) sebagai kunci bucket default.

  Sebagai contoh, jika kunci primer adalah (shop_id, user_id, ds) dan kunci partisi adalah ds, serta tidak ada kunci bucket yang ditentukan, kunci bucket default-nya adalah 'bucket.key' = 'user_id,shop_id'.

• Catatan dengan nilai hash yang sama ditetapkan ke bucket yang sama, yang memfasilitasi pemrosesan paralel dan penyeimbangan beban.

Untuk detail lebih lanjut, lihat Data bucketing.

Pembaruan parsial

Fluss mendukung pembaruan parsial untuk tabel kunci primer. Anda dapat menulis subset kolom untuk memperbarui catatan secara inkremental. Pembaruan parsial ini pada akhirnya digabungkan untuk membentuk baris lengkap.

CREATE TABLE T (
  k INT,
  v1 DOUBLE,
  v2 STRING,
  PRIMARY KEY (k) NOT ENFORCED
);

Anda dapat menulis kolom parsial dalam operasi terpisah, tetapi setiap penulisan harus mencakup kolom kunci primer, seperti yang ditunjukkan pada gambar berikut.

Mesin penggabungan (Merge engines)

Fluss Merge Engine adalah komponen inti untuk memproses pembaruan data secara efisien di tabel kunci primer. Komponen ini memungkinkan Anda menentukan cara catatan baru digabungkan dengan catatan yang sudah ada yang memiliki kunci primer yang sama. Anda juga dapat menentukan mesin penggabungan berbeda untuk menyesuaikan perilaku penggabungan sesuai kasus penggunaan Anda.

Mesin penggabungan berikut didukung:

• LastRow Merge Engine

  Jika catatan baru memiliki kunci primer yang sama dengan catatan yang sudah ada, mesin ini menyimpan catatan terbaru (berdasarkan waktu pemrosesan, yaitu catatan yang ditulis terakhir).

• FirstRow Merge Engine

  • Berbeda dengan strategi default, mesin ini menyimpan catatan paling awal (berdasarkan waktu pemrosesan, yaitu catatan yang ditulis pertama kali).

  • Mesin ini berguna untuk skenario di mana Anda perlu mempertahankan keadaan awal data.

• Versioned Merge Engine

  • Mesin ini menggabungkan catatan berdasarkan nomor versi.

  • Anda dapat menentukan versi catatan mana yang akan disimpan berdasarkan bidang versi.

  • Mesin ini ideal untuk skenario yang memerlukan pengendalian versi data atau pembaruan inkremental.

• Aggregation Merge Engine

  • Mengelompokkan catatan berdasarkan kunci primer yang sama dan menerapkan fungsi agregat tertentu pada setiap bidang non-kunci primer.

  • Mesin ini mencakup banyak fungsi agregat umum, sehingga menyederhanakan skenario agregasi data.

Generasi changelog

Fluss mencatat penyisipan, penghapusan, dan pembaruan data di tabel kunci primer sebagai changelog. Konsumen downstream dapat langsung membaca changelog ini untuk melacak perubahan tabel.

Sebagai contoh, jika Anda pertama kali menyisipkan dua catatan (1, 2.0, 'apple') dan (1, 4.0, 'banana') ke dalam tabel kunci primer Fluss, lalu menghapus catatan (1, 4.0, 'banana'), data changelog berikut dihasilkan:

+I(1, 2.0, 'apple')
-U(1, 2.0, 'apple')
+U(1, 4.0, 'banana')
-D(1, 4.0, 'banana')

Penyimpanan lokal dan siklus hidup partisi

Struktur penyimpanan tabel kunci primer terdiri dari LogTablet untuk aliran retraction dan KvTablet untuk pencarian state. Karena komponen-komponen ini menggunakan mekanisme berbeda, mengandalkan TTL standar saja tidak dapat sepenuhnya membebaskan ruang disk lokal.

• Manajemen LogTablet: Parameter table.log.ttl hanya mengontrol periode retensi changelog dan tidak memengaruhi ukuran penyimpanan KvTablet.

• Manajemen KvTablet (Direkomendasikan): Untuk sepenuhnya menghapus data kedaluwarsa dan membebaskan ruang disk lokal yang ditempati oleh KvTablet, Anda harus mengaktifkan Partition TTL.

  • Aktifkan auto-partition: Tetapkan table.auto-partition.enabled = true.

  • Tetapkan kebijakan retensi: Konfigurasikan table.auto-partition.num-retention untuk menentukan jumlah partisi yang dipertahankan.

  • Logika eksekusi: Sistem secara berkala memeriksa dan menghapus secara fisik partisi lama yang melebihi jumlah retensi, sehingga secara efektif membebaskan ruang penyimpanan lokal.

Kolom auto-increment

Kolom auto-increment secara otomatis menghasilkan nilai numerik unik yang meningkat untuk setiap baris, sehingga Anda tidak perlu menentukannya secara manual. Kasus penggunaan umum meliputi:

• Mempercepat penghitungan distinct: Memetakan identifier tipe STRING ke integer auto-increment. Melakukan penghitungan distinct pada nilai integer dapat meningkatkan kecepatan kueri hingga beberapa kali lipat.

• Agregasi RoaringBitmap: Gunakan fungsi bawaan rbm32 (32-bit) dan rbm64 (64-bit) di Aggregation Merge Engine untuk mengagregasi ID integer auto-increment ke dalam RoaringBitmap guna penghitungan distinct yang sangat efisien.

CREATE TABLE uid_mapping (
  user_id STRING,
  uid BIGINT,
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'auto-increment.fields' = 'uid',
  'bucket.num' = '2'
);

Fitur dasar

• Keunikan: Nilai yang dihasilkan oleh kolom auto-increment bersifat unik di seluruh tabel.

• Monotonisitas: Setiap bucket tabel menyimpan cache batch ID auto-increment di TabletServer. Akibatnya, nilai-nilai tersebut tidak dijamin monotonik secara global, hanya kira-kira meningkat seiring waktu.

  Ukuran cache dikontrol oleh properti tabel table.auto-increment.cache-size, dengan nilai default 100.000. Cache yang lebih besar meningkatkan performa alokasi tetapi mengurangi monotonisitas. Properti ini tidak dapat diubah setelah tabel dibuat.

Contoh

• table.auto-increment.cache-size = 100000 (default)

• Tabel memiliki 2 bucket. Saat sistem dimulai, sistem pre-allocates rentang ID untuk setiap bucket:

  • Bucket 0 → [1, 100000]

  • Bucket 1 → [100001, 200000]

Jika data ditulis ke bucket 1, uid dimulai dari 100001. Setelah bucket 0 menghabiskan 100.000 ID-nya, bucket tersebut meminta batch berikutnya, [200001, 300000], dan seterusnya. Tidak ada batasan jumlah total ID yang dapat dialokasikan setiap bucket. 100000 bukan batas atas; ini adalah ukuran batch cache.

Kasus penggunaan

Anda dapat menggunakan kolom auto-increment dengan opsi lookup.insert-if-not-exists dalam Flink Lookup Join untuk secara otomatis membangun tabel kamus selama pemrosesan aliran. Ketika kunci lookup tidak ditemukan, Fluss secara otomatis menyisipkan baris baru dan menetapkan ID auto-increment yang unik.

Kasus penggunaan umum adalah memetakan identifier string dengan kardinalitas tinggi ke ID integer ringkas untuk agregasi penghitungan distinct berikutnya yang efisien menggunakan RoaringBitmap. Untuk detail lebih lanjut, lihat dimension table join.

Batasan

• Hanya didukung untuk tabel kunci primer.

• Anda tidak dapat secara eksplisit menentukan nilai untuk kolom auto-increment; nilai tersebut harus ditetapkan secara implisit oleh sistem.

• Satu tabel hanya dapat memiliki satu kolom auto-increment.

• Tipe data kolom auto-increment harus berupa INT atau BIGINT.

• Anda tidak dapat menentukan nilai awal atau langkah increment untuk kolom auto-increment.

Membangun tabel kamus dengan Lookup Join

Dengan menggabungkan kolom auto-increment dan opsi lookup.insert-if-not-exists dari Flink Lookup Join, Anda dapat secara otomatis membangun tabel kamus selama pemrosesan aliran. Ketika kunci lookup tidak ditemukan, Fluss secara otomatis menyisipkan baris baru dan menetapkan ID auto-increment. Pendekatan ini sangat berguna untuk memetakan identifier string dengan kardinalitas tinggi ke ID integer ringkas guna deduplikasi dan agregasi efisien menggunakan RoaringBitmap. Untuk informasi dan contoh lebih lanjut, lihat dimension table join.

Membaca dan mengkueri tabel kunci primer

• Pembersihan data: Layanan Fluss hanya mengelola sumber daya penyimpanannya sendiri dan tidak melakukan operasi pembersihan data atau pengumpulan sampah di lapisan penyimpanan Paimon.

• Batasan kueri real-time: Fluss hanya menyediakan layanan kueri untuk data dalam siklus hidup yang dikonfigurasi. Setelah data melebihi periode retensinya, sistem menghapusnya dari lapisan Fluss, dan pengguna tidak dapat mengambilnya melalui antarmuka Fluss.

• Akses data historis: Di masa depan, Anda akan dapat mengkueri data partisi yang kedaluwarsa melalui mekanisme Union Read Paimon untuk mengakses dataset historis lengkap.