Realtime Compute for Apache Flink：MySQL

Informasi latar belakang

Konektor MySQL mendukung semua database yang kompatibel dengan protokol MySQL, seperti RDS MySQL, PolarDB for MySQL, OceanBase (mode MySQL), dan MySQL yang dikelola sendiri.

Fitur

Tabel sumber MySQL Change Data Capture (CDC) pertama-tama membaca seluruh data historis database, lalu beralih secara mulus ke pembacaan event binary log (binlog). Proses ini menjamin semantik tepat-sekali (exactly-once semantics), artinya tidak ada data yang terlewat atau diduplikasi, bahkan jika terjadi kegagalan. Tabel sumber CDC MySQL mendukung pembacaan data lengkap secara konkuren serta menerapkan pembacaan tanpa penguncian (lock-free) dan transfer data yang dapat dilanjutkan (resumable) menggunakan algoritma snapshot inkremental. Untuk informasi selengkapnya, lihat Tentang tabel sumber MySQL CDC.

Prasyarat

Sebelum menggunakan tabel sumber MySQL CDC, Anda harus mengonfigurasi database MySQL seperti yang dijelaskan dalam Konfigurasi MySQL. Konfigurasi berikut wajib dilakukan.

Batasan

Catatan

• Konfigurasikan secara eksplisit Server ID yang berbeda untuk setiap sumber data MySQL CDC

  Tujuan Server ID

  Anda harus mengonfigurasikan secara eksplisit Server ID yang berbeda untuk setiap sumber data MySQL CDC. Jika beberapa sumber data MySQL CDC berbagi Server ID yang sama dan tidak dapat menggunakan kembali koneksi, offset binlog akan menjadi tidak teratur. Hal ini mengakibatkan pembacaan berlebih atau kurang.

  Konfigurasi Server ID dalam berbagai skenario

  Anda dapat menentukan Server ID dalam pernyataan DDL, tetapi kami menyarankan agar Anda mengonfigurasinya menggunakan dynamic hints daripada dalam opsi DDL.

  • Paralelisme = 1 atau snapshot inkremental dinonaktifkan

    ## Tentukan Server ID spesifik saat framework snapshot inkremental tidak diaktifkan atau paralelisme adalah 1.
    SELECT * FROM source_table /*+ OPTIONS('server-id'='123456') */ ;

  • Paralelisme > 1 dan snapshot inkremental diaktifkan

    ## Tentukan rentang Server ID. Pastikan jumlah Server ID yang tersedia dalam rentang tersebut minimal sama dengan tingkat paralelisme. Misalnya, paralelisme adalah 3.
    SELECT * FROM source_table /*+ OPTIONS('server-id'='123456-123458') */ ;

  • Sinkronisasi data dengan CTAS

    Saat Anda menggunakan CTAS untuk sinkronisasi data dan sumber data CDC memiliki konfigurasi identik, sumber data akan digunakan kembali secara otomatis. Dalam kasus ini, Anda dapat mengonfigurasi Server ID yang sama untuk beberapa sumber data CDC. Untuk informasi selengkapnya, lihat Contoh 4: Beberapa pernyataan CTAS.

  • Beberapa tabel sumber non-CTAS yang tidak dapat digunakan kembali

    Jika pekerjaan berisi beberapa tabel sumber MySQL CDC dan tidak menggunakan pernyataan CTAS untuk sinkronisasi, sumber data tidak dapat digunakan kembali. Dalam kasus ini, Anda harus memberikan Server ID yang berbeda untuk setiap tabel sumber CDC. Demikian pula, jika framework snapshot inkremental diaktifkan dan paralelisme lebih besar dari 1, Anda harus menentukan rentang Server ID.

    select * from 
      source_table1 /*+ OPTIONS('server-id'='123456-123457') */
    left join 
      source_table2 /*+ OPTIONS('server-id'='123458-123459') */
    on source_table1.id=source_table2.id;

• Tabel sink

  • Jangan deklarasikan primary key auto-increment dalam pernyataan DDL. MySQL akan mengisi bidang ini secara otomatis saat menulis data.

  • Anda harus mendeklarasikan setidaknya satu bidang non-primary key. Jika tidak, akan terjadi error.

  • NOT ENFORCED dalam pernyataan DDL menunjukkan bahwa Flink tidak menegakkan pemeriksaan validitas pada primary key. Anda harus memastikan kebenaran dan integritas primary key. Untuk informasi selengkapnya, lihat Pemeriksaan Validitas.

• Tabel dimensi

  Untuk menggunakan indeks guna mempercepat kueri, urutan bidang dalam klausa JOIN harus sesuai dengan urutan definisi indeks. Ini dikenal sebagai aturan prefiks paling kiri (leftmost prefix rule). Misalnya, jika indeks berada pada (a, b, c), kondisi JOIN harus berupa ON t.a = x AND t.b = y.

  SQL yang dihasilkan Flink mungkin ditulis ulang oleh pengoptimal, sehingga mencegah penggunaan indeks selama kueri database aktual. Untuk memverifikasi apakah indeks digunakan, Anda dapat memeriksa rencana eksekusi (EXPLAIN) atau log kueri lambat di MySQL untuk melihat pernyataan SELECT aktual yang dieksekusi.

Contoh

• Tabel sumber CDC

  CREATE TEMPORARY TABLE mysqlcdc_source (
     order_id INT,
     order_date TIMESTAMP(0),
     customer_name STRING,
     price DECIMAL(10, 5),
     product_id INT,
     order_status BOOLEAN,
     PRIMARY KEY(order_id) NOT ENFORCED
  ) WITH (
    'connector' = 'mysql',
    'hostname' = '<yourHostname>',
    'port' = '3306',
    'username' = '<yourUsername>',
    'password' = '<yourPassword>',
    'database-name' = '<yourDatabaseName>',
    'table-name' = '<yourTableName>'
  );
  
  CREATE TEMPORARY TABLE blackhole_sink(
    order_id INT,
    customer_name STRING
  ) WITH (
    'connector' = 'blackhole'
  );
  
  INSERT INTO blackhole_sink
  SELECT order_id, customer_name FROM mysqlcdc_source;

• Tabel dimensi

  CREATE TEMPORARY TABLE datagen_source(
    a INT,
    b BIGINT,
    c STRING,
    `proctime` AS PROCTIME()
  ) WITH (
    'connector' = 'datagen'
  );
  
  CREATE TEMPORARY TABLE mysql_dim (
    a INT,
    b VARCHAR,
    c VARCHAR
  ) WITH (
    'connector' = 'mysql',
    'hostname' = '<yourHostname>',
    'port' = '3306',
    'username' = '<yourUsername>',
    'password' = '<yourPassword>',
    'database-name' = '<yourDatabaseName>',
    'table-name' = '<yourTableName>'
  );
  
  CREATE TEMPORARY TABLE blackhole_sink(
    a INT,
    b STRING
  ) WITH (
    'connector' = 'blackhole'
  );
  
  INSERT INTO blackhole_sink
  SELECT T.a, H.b
  FROM datagen_source AS T JOIN mysql_dim FOR SYSTEM_TIME AS OF T.`proctime` AS H ON T.a = H.a;

• Tabel sink

  CREATE TEMPORARY TABLE datagen_source (
    `name` VARCHAR,
    `age` INT
  ) WITH (
    'connector' = 'datagen'
  );
  
  CREATE TEMPORARY TABLE mysql_sink (
    `name` VARCHAR,
    `age` INT
  ) WITH (
    'connector' = 'mysql',
    'hostname' = '<yourHostname>',
    'port' = '3306',
    'username' = '<yourUsername>',
    'password' = '<yourPassword>',
    'database-name' = '<yourDatabaseName>',
    'table-name' = '<yourTableName>'
  );
  
  INSERT INTO mysql_sink
  SELECT * FROM datagen_source;

• Sumber data ingestion

  source:
    type: mysql
    name: MySQL Source
    hostname: ${mysql.hostname}
    port: ${mysql.port}
    username: ${mysql.username}
    password: ${mysql.password}
    tables: ${mysql.source.table}
    server-id: 7601-7604
  
  sink:
    type: values
    name: Values Sink
    print.enabled: true
    sink.print.logger: true

Tentang tabel sumber MySQL CDC

• Cara kerja

  Tabel sumber MySQL CDC memindai tabel lengkap saat startup dan membaginya menjadi beberapa chunk berdasarkan primary key. Sistem mencatat offset binlog saat ini lalu menggunakan algoritma snapshot inkremental untuk membaca data setiap chunk menggunakan pernyataan SELECT. Pekerjaan melakukan checkpoint berkala untuk mencatat chunk yang telah selesai. Jika terjadi failover, pekerjaan melanjutkan membaca hanya chunk yang belum selesai. Setelah semua chunk terbaca, pekerjaan membaca record perubahan inkremental dari offset binlog yang sebelumnya dicatat. Pekerjaan Flink terus melakukan checkpoint berkala untuk mencatat offset binlog. Jika terjadi failover, pekerjaan melanjutkan pemrosesan dari offset binlog terakhir yang dicatat. Proses ini mencapai semantik tepat-sekali (exactly-once semantics).

  Untuk detail lebih lanjut tentang algoritma snapshot inkremental, lihat Konektor MySQL CDC.

• Metadata

  Metadata sangat berguna dalam skenario penggabungan database dan tabel yang di-shard. Setelah digabung, bisnis sering kali masih perlu mengidentifikasi database dan tabel sumber untuk setiap baris data. Kolom metadata memungkinkan Anda mengakses informasi ini. Oleh karena itu, Anda dapat dengan mudah menggabungkan beberapa tabel yang di-shard menjadi satu tabel tujuan menggunakan kolom metadata.

  Sumber MySQL CDC mendukung sintaks kolom metadata. Anda dapat mengakses metadata berikut melalui kolom metadata.

  Kunci metadata	Tipe metadata	Deskripsi
  database_name	STRING NOT NULL	Nama database yang berisi baris tersebut.
  table_name	STRING NOT NULL	Nama tabel yang berisi baris tersebut.
  op_ts	TIMESTAMP_LTZ(3) NOT NULL	Waktu saat baris tersebut diubah di database. Jika record berasal dari data historis yang sudah ada di tabel, bukan dari binlog, nilai ini selalu 0.
  op_type	STRING NOT NULL	Jenis perubahan baris tersebut. +I: Pesan INSERT -D: Pesan DELETE -U: Pesan UPDATE_BEFORE +U: Pesan UPDATE_AFTER Catatan Hanya didukung di Ververica Runtime (VVR) 8.0.7 dan versi selanjutnya.
  query_log	STRING NOT NULL	Baca record log kueri MySQL yang sesuai dengan baris ini. Catatan MySQL harus memiliki parameter binlog_rows_query_log_events diaktifkan untuk mencatat log kueri.

  Contoh berikut menunjukkan cara menggabungkan beberapa tabel orders dari berbagai database yang di-shard dalam instans MySQL dan menyinkronkannya ke tabel holo_orders di Hologres.

  CREATE TEMPORARY TABLE mysql_orders (
    db_name STRING METADATA FROM 'database_name' VIRTUAL,  -- Baca nama database.
    table_name STRING METADATA  FROM 'table_name' VIRTUAL, -- Baca nama tabel.
    operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL, -- Baca timestamp perubahan.
    op_type STRING METADATA FROM 'op_type' VIRTUAL, -- Baca jenis perubahan.
    order_id INT,
    order_date TIMESTAMP(0),
    customer_name STRING,
    price DECIMAL(10, 5),
    product_id INT,
    order_status BOOLEAN,
    PRIMARY KEY(order_id) NOT ENFORCED
  ) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '3306',
    'username' = 'flinkuser',
    'password' = 'flinkpw',
    'database-name' = 'mydb_.*', -- Cocokkan beberapa database yang di-shard menggunakan ekspresi reguler.
    'table-name' = 'orders_.*'   -- Cocokkan beberapa tabel yang di-shard menggunakan ekspresi reguler.
  );
  
  INSERT INTO holo_orders SELECT * FROM mysql_orders;

  Berdasarkan kode di atas, jika Anda mengatur scan.read-changelog-as-append-only.enabled ke true dalam klausa WITH, output bervariasi berdasarkan konfigurasi primary key tabel downstream:

  • Jika primary key tabel downstream adalah order_id, output hanya berisi perubahan terakhir untuk setiap primary key di tabel upstream. Misalnya, jika perubahan terakhir untuk primary key adalah operasi penghapusan, Anda akan melihat record di tabel downstream dengan primary key yang sama dan op_type -D.

  • Jika primary key tabel downstream adalah order_id, operation_ts, dan op_type, output berisi riwayat perubahan lengkap untuk setiap primary key di tabel upstream.

• Dukungan ekspresi reguler

  Tabel sumber MySQL CDC mendukung penggunaan ekspresi reguler dalam nama tabel atau nama database untuk mencocokkan beberapa tabel atau database. Contoh berikut menunjukkan cara menentukan beberapa tabel menggunakan ekspresi reguler.

  CREATE TABLE products (
    db_name STRING METADATA FROM 'database_name' VIRTUAL,
    table_name STRING METADATA  FROM 'table_name' VIRTUAL,
    operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
    order_id INT,
    order_date TIMESTAMP(0),
    customer_name STRING,
    price DECIMAL(10, 5),
    product_id INT,
    order_status BOOLEAN,
    PRIMARY KEY(order_id) NOT ENFORCED
  ) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '3306',
    'username' = 'root',
    'password' = '123456',
    'database-name' = '(^(test).*|^(tpc).*|txc|.*[p$]|t{2})', -- Cocokkan beberapa database menggunakan ekspresi reguler.
    'table-name' = '(t[5-8]|tt)' -- Cocokkan beberapa tabel menggunakan ekspresi reguler.
  );

  Penjelasan ekspresi reguler dalam contoh di atas:

  • ^(test).* adalah contoh pencocokan awalan. Ekspresi ini mencocokkan nama database yang dimulai dengan test, seperti test1 atau test2.

  • .*[p$] adalah contoh pencocokan akhiran. Ekspresi ini mencocokkan nama database yang diakhiri dengan p, seperti cdcp atau edcp.

  • txc adalah pencocokan eksak. Ini mencocokkan nama database spesifik txc.

  Saat mencocokkan nama tabel yang memenuhi syarat penuh, MySQL CDC menggunakan nama database dan nama tabel untuk mengidentifikasi tabel secara unik. Sistem menggunakan pola database-name.table-name untuk pencocokan. Misalnya, pola (^(test).*|^(tpc).*|txc|.*[p$]|t{2}).(t[5-8]|tt) mencocokkan tabel seperti txc.tt dan test2.test5 dalam database.

  Penting

  Dalam konfigurasi pekerjaan SQL, opsi table-name dan database-name tidak mendukung penggunaan koma (,) untuk menentukan beberapa tabel atau database.

  • Untuk mencocokkan beberapa tabel atau menggunakan beberapa ekspresi reguler, Anda dapat memisahkannya dengan GARIS TEGAK (|) dan membungkusnya dalam tanda kurung. Misalnya, untuk membaca tabel user dan product, Anda dapat mengonfigurasi table-name sebagai (user|product).

  • Jika ekspresi reguler berisi koma, Anda harus menulis ulang menggunakan operator GARIS TEGAK (|). Misalnya, ekspresi reguler mytable_\d{1, 2} harus ditulis ulang menjadi yang setara (mytable_\d{1}|mytable_\d{2}) untuk menghindari penggunaan koma.

• Kontrol konkurensi

  Konektor MySQL mendukung pembacaan data lengkap secara konkuren, yang meningkatkan efisiensi pemuatan data. Saat dikombinasikan dengan Autopilot di konsol Realtime Compute for Apache Flink, konektor secara otomatis mengurangi skala selama fase inkremental setelah pembacaan konkuren selesai. Hal ini menghemat sumber daya komputasi.

  Di Konsol pengembangan Realtime Compute, Anda dapat mengatur paralelisme pekerjaan di halaman Konfigurasi Sumber Daya baik dalam mode Dasar maupun mode Ahli. Perbedaannya sebagai berikut:

  • Paralelisme yang diatur dalam mode Dasar adalah paralelisme global untuk seluruh pekerjaan.

  • Mode Ahli memungkinkan Anda mengatur paralelisme untuk VERTEX tertentu sesuai kebutuhan.

  Untuk informasi selengkapnya tentang konfigurasi sumber daya, lihat Konfigurasi informasi penerapan pekerjaan.

  Penting

  Terlepas dari apakah Anda menggunakan mode Dasar atau mode Ahli, rentang server-id yang dideklarasikan dalam tabel harus lebih besar dari atau sama dengan paralelisme pekerjaan. Misalnya, jika rentang server-id adalah 5404-5412, terdapat delapan ID server unik. Oleh karena itu, pekerjaan dapat memiliki maksimal delapan tugas paralel. Pekerjaan berbeda untuk instans MySQL yang sama harus memiliki rentang server-id yang tidak tumpang tindih. Setiap pekerjaan harus secara eksplisit mengonfigurasi server-id yang berbeda.

• Autopilot Auto Scale-in

  Fase data lengkap mengakumulasi banyak data historis. Untuk meningkatkan efisiensi pembacaan, data historis biasanya dibaca secara konkuren. Namun, dalam fase binlog inkremental, satu konkurensi biasanya sudah cukup karena volume data binlog kecil dan urutan global harus dipertahankan. Autopilot secara otomatis menyeimbangkan kinerja dan sumber daya untuk memenuhi persyaratan berbeda antara fase lengkap dan inkremental.

  Autopilot memantau trafik untuk setiap tugas di Sumber CDC MySQL. Saat memasuki fase binlog, jika hanya satu tugas yang menangani pembacaan binlog sementara yang lain menganggur, Autopilot secara otomatis mengurangi jumlah CU dan paralelisme Sumber. Untuk mengaktifkan Autopilot, atur mode Autopilot ke Aktif di halaman Operasi dan Pemeliharaan pekerjaan.

  Catatan

  Interval pemicu minimum default untuk penskalaan turun paralelisme adalah 24 jam. Untuk informasi selengkapnya tentang parameter dan detail Autopilot, lihat Konfigurasi Autopilot.

• Startup Mode

  Anda dapat menggunakan opsi scan.startup.mode untuk menentukan mode startup untuk tabel sumber MySQL CDC. Nilai yang valid dijelaskan sebagai berikut:

  • initial (default): Melakukan pembacaan lengkap tabel database saat startup pertama kali, lalu beralih ke mode inkremental untuk membaca binlog.

  • earliest-offset: Melewati fase snapshot dan mulai membaca dari offset binlog paling awal yang tersedia.

  • latest-offset: Melewati fase snapshot dan mulai membaca dari akhir binlog. Dalam mode ini, tabel sumber hanya membaca perubahan yang terjadi setelah pekerjaan dimulai.

  • specific-offset: Melewati fase snapshot dan mulai membaca dari offset binlog tertentu. Anda dapat menentukan offset berdasarkan nama file binlog dan posisi atau berdasarkan set GTID.

  • timestamp: Melewati fase snapshot dan mulai membaca event binlog dari timestamp tertentu.

  Contoh:

  CREATE TABLE mysql_source (...) WITH (
      'connector' = 'mysql-cdc',
      'scan.startup.mode' = 'earliest-offset', -- Mulai dari offset paling awal.
      'scan.startup.mode' = 'latest-offset', -- Mulai dari offset terbaru.
      'scan.startup.mode' = 'specific-offset', -- Mulai dari offset tertentu.
      'scan.startup.mode' = 'timestamp', -- Mulai dari timestamp tertentu.
      'scan.startup.specific-offset.file' = 'mysql-bin.000003', -- Tentukan nama file binlog untuk mode specific-offset.
      'scan.startup.specific-offset.pos' = '4', -- Tentukan posisi binlog untuk mode specific-offset.
      'scan.startup.specific-offset.gtid-set' = '24DA167-0C0C-11E8-8442-00059A3C7B00:1-19', -- Tentukan set GTID untuk mode specific-offset.
      'scan.startup.timestamp-millis' = '1667232000000' -- Tentukan timestamp startup untuk mode timestamp.
      ...
  )

  Penting

  • Sumber MySQL mencatat posisi saat ini pada level INFO selama checkpoint. Awalan log adalah Binlog offset on checkpoint {checkpoint-id}. Log ini membantu Anda memulai ulang pekerjaan dari posisi checkpoint tertentu.

  • Jika skema tabel yang sedang dibaca telah berubah di masa lalu, memulai dari earliest-offset, specific-offset, atau timestamp dapat menyebabkan error. Hal ini karena pembaca Debezium secara internal menyimpan skema terbaru, dan data lama dengan skema yang tidak cocok tidak dapat diurai dengan benar.

• Tabel sumber CDC tanpa kunci

  • Untuk menggunakan tabel tanpa kunci, Anda harus mengatur scan.incremental.snapshot.chunk.key-column dan memilih hanya bidang non-null.

  • Semantik pemrosesan tabel sumber CDC tanpa kunci bergantung pada perilaku kolom yang ditentukan dalam scan.incremental.snapshot.chunk.key-column:

    • Jika kolom yang ditentukan tidak pernah diperbarui, semantik tepat-sekali dijamin.

    • Jika kolom yang ditentukan diperbarui, hanya semantik at-least-once yang dijamin. Namun, Anda dapat memastikan kebenaran data dengan menggabungkannya dengan sistem downstream, menentukan primary key downstream, dan menggunakan operasi idempoten.

• Baca log cadangan dari RDS MySQL

  Tabel sumber MySQL CDC mendukung pembacaan log cadangan dari Alibaba Cloud RDS MySQL. Fitur ini sangat berguna ketika fase snapshot lengkap memakan waktu lama. Dalam kasus ini, file binlog lokal mungkin secara otomatis dibersihkan, sedangkan file cadangan yang diunggah secara manual atau otomatis masih ada.

  Contoh:

  CREATE TABLE mysql_source (...) WITH (
      'connector' = 'mysql-cdc',
      'rds.region-id' = 'cn-beijing',
      'rds.access-key-id' = 'xxxxxxxxx', 
      'rds.access-key-secret' = 'xxxxxxxxx', 
      'rds.db-instance-id' = 'rm-xxxxxxxxxxxxxxxxx', 
      'rds.main-db-id' = '12345678',
      'rds.download.timeout' = '60s'
      ...
  )

• Aktifkan penggunaan ulang sumber CDC

  Satu pekerjaan dengan beberapa tabel sumber MySQL CDC meluncurkan beberapa klien binlog. Jika semua tabel sumber membaca dari instans MySQL yang sama, praktik ini meningkatkan tekanan pada database. Untuk informasi selengkapnya, lihat FAQ MySQL CDC.

  Solusi

  VVR 8.0.7 dan versi selanjutnya mendukung penggunaan ulang sumber MySQL CDC. Penggunaan ulang menggabungkan tabel sumber MySQL CDC yang kompatibel. Penggabungan terjadi ketika tabel sumber memiliki konfigurasi identik kecuali nama database, nama tabel, dan server-id. Mesin secara otomatis menggabungkan sumber MySQL CDC dalam pekerjaan yang sama.

  Prosedur

  1. Anda dapat menggunakan perintah SET dalam pekerjaan SQL:

     SET 'table.optimizer.source-merge.enabled' = 'true';
     
     # (Versi VVR 8.0.8 dan 8.0.9) Atur juga ini:
     SET 'sql-gateway.exec-plan.enabled' = 'false';

     VVR 11.1 dan versi selanjutnya mengaktifkan penggunaan ulang secara default.

  2. Jalankan pekerjaan tanpa status. Memodifikasi konfigurasi penggunaan ulang mengubah topologi pekerjaan. Anda harus menjalankan pekerjaan tanpa status. Jika tidak, pekerjaan mungkin gagal dijalankan atau kehilangan data. Jika sumber digabung, Anda dapat melihat node MergetableSourceScan dalam topologi pekerjaan.

  Penting

  • Setelah Anda mengaktifkan penggunaan ulang, jangan atur pipeline.operator-chaining ke false. Menonaktifkan operator chaining menambahkan overhead serialisasi dan deserialisasi. Semakin banyak sumber yang digabung, semakin besar overhead-nya.

  • Di VVR 8.0.7, menonaktifkan operator chaining menyebabkan masalah serialisasi.

Percepat pembacaan binlog

Saat Anda menggunakan konektor MySQL sebagai tabel sumber atau sumber data ingestion, sistem mengurai file binlog untuk menghasilkan berbagai pesan perubahan selama fase inkremental. File binlog mencatat semua perubahan tabel dalam format biner. Anda dapat mempercepat penguraian file binlog dengan cara berikut:

• Aktifkan konfigurasi filter penguraian

  • Gunakan opsi scan.only.deserialize.captured.tables.changelog.enabled untuk mengurai event perubahan hanya untuk tabel yang ditentukan.

• Optimalkan opsi Debezium

  debezium.max.queue.size: 162580
  debezium.max.batch.size: 40960
  debezium.poll.interval.ms: 50

  • debezium.max.queue.size: Jumlah maksimum record yang dapat ditampung oleh antrian pemblokiran. Saat Debezium membaca aliran event dari database, sistem menempatkan event dalam antrian pemblokiran sebelum menuliskannya ke downstream. Nilai default adalah 8192.

  • debezium.max.batch.size: Jumlah maksimum event yang diproses per iterasi. Nilai default adalah 2048.

  • debezium.poll.interval.ms: Jumlah milidetik yang ditunggu konektor sebelum meminta event perubahan baru. Nilai default adalah 1000 ms (1 detik).

Contoh:

CREATE TABLE mysql_source (...) WITH (
    'connector' = 'mysql-cdc',
    -- Konfigurasi Debezium
    'debezium.max.queue.size' = '162580',
    'debezium.max.batch.size' = '40960',
    'debezium.poll.interval.ms' = '50',
    -- Aktifkan filter penguraian
    'scan.only.deserialize.captured.tables.changelog.enabled' = 'true',  -- Urai event perubahan hanya untuk tabel yang ditentukan.
    ...
)

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: ${mysql.source.table}
  server-id: 7601-7604
  # Konfigurasi Debezium
  debezium.max.queue.size: 162580
  debezium.max.batch.size: 40960
  debezium.poll.interval.ms: 50
  # Aktifkan filter penguraian
  scan.only.deserialize.captured.tables.changelog.enabled: true

Versi enterprise MySQL CDC mengonsumsi binlog dengan kecepatan 85 MB/detik, sekitar dua kali lipat kecepatan versi komunitas open-source. Jika laju pembuatan binlog melebihi 85 MB/detik (setara dengan satu file 512 MB setiap 6 detik), latensi pekerjaan Flink terus meningkat. Latensi secara bertahap berkurang setelah laju pembuatan binlog melambat. Saat file binlog berisi transaksi besar, latensi pemrosesan mungkin sementara meningkat lalu berkurang setelah log transaksi terbaca.

API DataStream MySQL CDC

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
public class MySqlSourceExample {
  public static void main(String[] args) throws Exception {
    MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
        .hostname("yourHostname")
        .port(yourPort)
        .databaseList("yourDatabaseName") // Atur database yang ditangkap.
        .tableList("yourDatabaseName.yourTableName") // Atur tabel yang ditangkap.
        .username("yourUsername")
        .password("yourPassword")
        .deserializer(new JsonDebeziumDeserializationSchema()) // Mengonversi SourceRecord ke String JSON.
        .build();
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // Aktifkan checkpointing.
    env.enableCheckpointing(3000);
    env
      .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
      // Atur 4 tugas sumber paralel.
      .setParallelism(4)
      .print().setParallelism(1); // Gunakan paralelisme 1 untuk sink guna mempertahankan urutan pesan.
    env.execute("Print MySQL Snapshot + Binlog");
  }
}

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-core</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-base</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-common</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-api-java-bridge</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>com.alibaba.ververica</groupId>
    <artifactId>ververica-connector-mysql</artifactId>
    <version>${vvr.version}</version>
</dependency>

FAQ

Untuk informasi tentang masalah yang mungkin Anda temui saat menggunakan tabel sumber CDC, lihat Masalah CDC.