Hologres:Konsumsi data binlog Hologres dengan Flink

Sebelum mengonsumsi data binlog Hologres, perhatikan hal-hal berikut:

• Anda hanya dapat mengonsumsi data binlog dari Hologres V0.9 atau versi yang lebih baru. Anda hanya dapat mengonfigurasi daftar putih engine untuk Hologres V1.3.21 atau versi yang lebih baru. Jika Anda mengaktifkan daftar putih pada instans yang menjalankan versi sebelum Hologres V1.3.21, konsumsi binlog akan gagal. Jika versi instans Anda lebih lama dari yang disyaratkan, Anda dapat Mendapatkan dukungan online untuk Hologres untuk bantuan.

• Hologres mendukung fitur binlog di tingkat tabel untuk tabel berorientasi baris dan tabel berorientasi kolom. Mulai dari Hologres V1.1, tabel penyimpanan hibrida baris-kolom juga didukung. Setelah Anda mengaktifkan fitur binlog, secara teoretis tabel berorientasi kolom memiliki overhead yang lebih tinggi dibandingkan tabel berorientasi baris. Oleh karena itu, untuk skenario dengan pembaruan data yang sering, kami menyarankan agar Anda mengaktifkan fitur binlog untuk tabel berorientasi baris.

• Untuk informasi lebih lanjut tentang dukungan binlog serta cara mengaktifkan dan mengonfigurasi fitur ini, lihat Berlangganan binlog Hologres.

• Hanya Realtime Compute for Apache Flink yang disediakan oleh Alibaba Cloud yang mendukung pengonsumsian data binlog Hologres. Dalam mode HoloHub, pengonsumsian data binlog Hologres dengan Flink hanya mendukung tipe data sederhana. Mulai dari Flink VVR 6.0.3, Anda dapat mengonsumsi data binlog Hologres dalam mode JDBC. Dibandingkan dengan mode HoloHub, mode JDBC mendukung lebih banyak tipe data. Untuk informasi selengkapnya, lihat Pemetaan tipe data antara Blink/Flink dan Hologres. Mode ini memerlukan izin tambahan. Untuk detailnya, lihat Izin.

• Anda tidak dapat mengonsumsi data binlog dari tabel partisi induk.

• Hologres V2.0 memberikan dukungan terbatas untuk mode HoloHub. Mulai dari V2.1, mode HoloHub ditinggalkan dan sepenuhnya digantikan oleh mode JDBC. Sebelum meningkatkan instans Hologres Anda, rujuk ke Beralih dari mode HoloHub ke mode JDBC untuk memeriksa Pekerjaan Flink Anda yang saat ini menggunakan mode HoloHub, tingkatkan Pekerjaan Flink VVR Anda sesuai, lalu tingkatkan instans Hologres Anda.

Dalam mode ini, data binlog yang dikonsumsi oleh Source diteruskan ke node downstream sebagai data Flink biasa. Artinya, semua data diperlakukan sebagai data bertipe Insert, dan Anda dapat memilih cara memproses data berdasarkan tipe tertentu hg_binlog_event_type sesuai kebutuhan bisnis Anda. Setelah binlog diaktifkan untuk tabel Hologres, Anda dapat menggunakan DDL berikut untuk tabel sumber (dalam mode non-CDC) di Flink guna mengonsumsi binlog secara real time.

create table test_message_src_binlog_table(
  hg_binlog_lsn BIGINT,
  hg_binlog_event_type BIGINT,
  hg_binlog_timestamp_us BIGINT,
  id INTEGER,
  title VARCHAR,
  body VARCHAR
) with (
  'connector'='hologres',
  'dbname'='<yourDbname>',
  'tablename'='<yourTablename>',
  'username'='<yourAccessID>',
  'password'='<yourAccessSecret>',
  'endpoint'='<yourEndpoint>',
  'binlog' = 'true',
  'binlogMaxRetryTimes' = '10',
  'binlogRetryIntervalMs' = '500',
  'binlogBatchReadSize' = '100'
);

• Tiga parameter binlogxxx merupakan bidang sistem binlog. Nama dan tipenya tetap dan tidak dapat diubah.

• Bidang lainnya sesuai dengan bidang yang Anda definisikan sendiri dan harus ditulis dalam huruf kecil semua.

Dalam mode ini, Source mengonsumsi data binlog dan secara otomatis menetapkan tipe Flink RowKind yang benar (INSERT, DELETE, UPDATE_BEFORE, atau UPDATE_AFTER) untuk setiap baris berdasarkan hg_binlog_event_type. Hal ini memungkinkan sinkronisasi cermin data tabel, mirip dengan fitur CDC di MySQL dan PostgreSQL.

Setelah Anda mengaktifkan fitur binlog untuk tabel Hologres, Anda dapat menggunakan DDL berikut di Flink untuk membuat tabel sumber guna mengonsumsi data binlog secara real time dalam mode CDC.

create table test_message_src_binlog_table(
  id INTEGER,
  title VARCHAR,
  body VARCHAR
) with (
  'connector'='hologres',
  'dbname'='<yourDbname>',
  'tablename'='<yourTablename>',
  'username'='<yourAccessID>',
  'password'='<yourAccessSecret>',
  'endpoint'='<yourEndpoint>',
  'binlog' = 'true',
  'cdcMode' = 'true',
  'binlogMaxRetryTimes' = '10',
  'binlogRetryIntervalMs' = '500',
  'binlogBatchReadSize' = '100'
);