All Products
Search

This Product

    Realtime Compute for Apache Flink:Flink CDC: Sinkronisasi seluruh database MySQL ke Kafka

    Document Center

    Realtime Compute for Apache Flink:Flink CDC: Sinkronisasi seluruh database MySQL ke Kafka

    Last Updated:Jun 21, 2026

    Topik ini menjelaskan cara menyinkronkan seluruh database MySQL ke Kafka. Pendekatan ini mengurangi beban yang ditimbulkan oleh beberapa job terhadap database MySQL.

    Latar Belakang

    Tabel sumber MySQL CDC menangkap data dari MySQL dan menyinkronkan perubahan real-time dari tabel tersebut. Hal ini umum dalam skenario komputasi kompleks, misalnya ketika sebuah tabel berfungsi sebagai tabel dimensi dalam operasi JOIN dengan tabel data lainnya. Satu tabel MySQL bisa menjadi dependensi bagi beberapa job. Ketika beberapa job memproses data dari tabel MySQL yang sama, database membuka beberapa koneksi, yang memberikan tekanan signifikan pada server dan jaringan MySQL.

    Cara Kerja

    Untuk mengurangi tekanan pada database MySQL hulu, Realtime Compute for Apache Flink dapat menyinkronkan seluruh database MySQL ke Kafka. Solusi ini memperkenalkan Kafka sebagai lapisan perantara, menggunakan job data ingestion Flink CDC untuk menyinkronkan data ke Kafka.

    Dalam satu job, data dari database MySQL hulu disinkronkan ke Kafka secara real-time. Setiap tabel MySQL ditulis ke topik Kafka yang sesuai dalam mode upsert. Job hilir kemudian menggunakan konektor upsert Kafka untuk membaca data dari topik-topik tersebut alih-alih mengakses tabel MySQL secara langsung. Metode ini secara efektif mengurangi tekanan yang ditimbulkan oleh beberapa job terhadap database MySQL.

    Gambar 1

    Batasan

    • Setiap tabel MySQL yang Anda sinkronkan harus memiliki primary key.

    • Anda dapat menggunakan cluster Kafka yang dikelola sendiri, cluster Kafka EMR, atau ApsaraMQ for Kafka. Jika Anda menggunakan ApsaraMQ for Kafka, Anda hanya dapat menghubungkannya melalui default endpoint.

    • Ruang penyimpanan kluster Kafka harus lebih besar daripada ukuran tabel sumber. Jika tidak, data dapat hilang karena ruang penyimpanan tidak mencukupi. Topik yang dibuat untuk sinkronisasi database adalah topik compacted. Dalam topik compacted, hanya pesan terbaru untuk setiap message key yang dipertahankan, tetapi data tidak pernah kedaluwarsa. Artinya, topik compacted menyimpan volume data yang kira-kira setara dengan ukuran tabel sumber.

    Contoh Skenario

    Sebagai contoh, dalam skenario analitik ulasan pesanan real-time, terdapat tiga tabel: tabel pengguna (user), tabel pesanan (order), dan tabel umpan balik pengguna (feedback). Data dalam tabel-tabel tersebut ditampilkan pada gambar berikut.mysql database

    Untuk menampilkan informasi pesanan pengguna dan ulasan pengguna, Anda perlu melakukan join tabel user untuk mengambil username dari field name. Contoh SQL berikut menunjukkan operasi ini.

    -- Join informasi pesanan dengan tabel user untuk menampilkan username dan nama produk untuk setiap pesanan.
SELECT order.id as order_id, product, user.name as user_name
FROM order LEFT JOIN user
ON order.user_id = user.id;
-- Join ulasan dengan tabel user untuk menampilkan isi setiap ulasan dan username yang sesuai.
SELECT feedback.id as feedback_id, comment, user.name as user_name
FROM feedback LEFT JOIN user
ON feedback.user_id = user.id;

    Untuk dua job SQL di atas, keduanya menggunakan tabel user. Saat runtime, kedua job membaca data full dan incremental dari MySQL. Pembacaan full memerlukan pembuatan koneksi MySQL, sedangkan pembacaan incremental memerlukan pembuatan client Binlog. Seiring bertambahnya jumlah job, permintaan terhadap sumber daya koneksi MySQL dan client Binlog juga meningkat, sehingga memberikan tekanan signifikan pada database hulu. Untuk meringankan tekanan ini, Anda dapat menggunakan job data ingestion Flink CDC untuk menyinkronkan data dari database MySQL hulu ke Kafka secara real-time guna dikonsumsi oleh beberapa job hilir.

    Prasyarat

    Persiapan

    Persiapkan sumber data MySQL

    1. Buat database ApsaraDB RDS for MySQL. Untuk informasi selengkapnya, lihat Buat database.

      Buat database bernama order_dw untuk instans target.

    2. Persiapkan sumber data MySQL CDC.

      1. Pada halaman detail instans, klik Log on to Database di bagian atas halaman.

      2. Pada kotak dialog login DMS yang muncul, masukkan username dan password untuk akun database yang telah Anda buat, lalu klik Login.

      3. Setelah login, klik ganda database order_dw di panel kiri untuk mengganti database.

      4. Pada SQL Console, masukkan statement DDL untuk membuat tiga tabel bisnis dan statement untuk memasukkan data.

        CREATE TABLE `user` (
  id bigint not null primary key,
  name varchar(50) not null
);
CREATE TABLE `order` (
  id bigint not null primary key,
  product varchar(50) not null,
  user_id bigint not null
);
CREATE TABLE `feedback` (
  id bigint not null primary key,
  user_id bigint not null,
  comment varchar(50) not null
);
-- Persiapkan data
INSERT INTO `user` VALUES(1, 'Tom'),(2, 'Jerry');
INSERT INTO `order` VALUES
(1, 'Football', 2),
(2, 'Basket', 1);
INSERT INTO `feedback` VALUES
(1, 1, 'Good.'),
(2, 2, 'Very good');

    3. Klik Execute, lalu klik Execute.

    Prosedur

    1. Buat dan mulai job data ingestion Flink CDC untuk menyinkronkan data dari database MySQL hulu ke Kafka secara real-time guna dikonsumsi oleh beberapa job hilir. Job sinkronisasi database secara otomatis membuat topik. Anda dapat menentukan nama topik menggunakan modul route. Topik menggunakan pengaturan default kluster Kafka untuk jumlah partisi dan replika, serta cleanup.policy diatur ke compact.

      Nama topik default

      Secara default, topik Kafka yang dibuat oleh job sinkronisasi database menggunakan format penamaan {database_name}.{table_name}. Job berikut membuat tiga topik: order_dw.user, order_dw.order, dan order_dw.feedback.

      1. Pada halaman Development > Data Ingestion, buat job data ingestion Flink CDC dan salin kode berikut ke editor YAML.

        source:
  type: mysql
  name: MySQL Source
  hostname: #{hostname}
  port: 3306
  username: #{usernmae}
  password: #{password}
  tables: order_dw.\.*
  server-id: 28601-28604
  # (Opsional) Sinkronkan data dari tabel yang baru dibuat selama fase inkremental.
  scan.binlog.newly-added-table.enabled: true
  # (Opsional) Sinkronkan komentar tabel dan field.
  include-comments.enabled: true
  # (Opsional) Utamakan pengiriman unbounded splits untuk menghindari potensi masalah OutOfMemory pada TaskManager.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Opsional) Aktifkan filter parsing untuk mempercepat pembacaan.
  scan.only.deserialize.captured.tables.changelog.enabled: true
sink:
  type: upsert-kafka
  name: upsert-kafka Sink
  properties.bootstrap.servers: xxxx.alikafka.aliyuncs.com:9092
  # Parameter berikut diperlukan untuk ApsaraMQ for Kafka.
  aliyun.kafka.accessKeyId: #{ak}
  aliyun.kafka.accessKeySecret: #{sk}
  aliyun.kafka.instanceId: #{instanceId}
  aliyun.kafka.endpoint: #{endpoint}
  aliyun.kafka.regionId: #{regionId}

      2. Di pojok kanan atas, klik Deploy untuk deploy job.

      3. Di bilah navigasi kiri, klik Operations Center > Deployments. Pada kolom Actions job target, klik Start, pilih Initial Mode, lalu klik Start.

      Nama topik per tabel

      Anda dapat menggunakan modul route untuk menentukan nama topik untuk setiap tabel. Job berikut membuat tiga topik: user1, order2, dan feedback3.

      1. Pada halaman Development > Data Ingestion, buat job data ingestion Flink CDC dan salin kode berikut ke editor YAML.

        source:
  type: mysql
  name: MySQL Source
  hostname: #{hostname}
  port: 3306
  username: #{usernmae}
  password: #{password}
  tables: order_dw.\.*
  server-id: 28601-28604
  # (Opsional) Sinkronkan data dari tabel yang baru dibuat selama fase inkremental.
  scan.binlog.newly-added-table.enabled: true
  # (Opsional) Sinkronkan komentar tabel dan field.
  include-comments.enabled: true
  # (Opsional) Utamakan pengiriman unbounded splits untuk menghindari potensi masalah OutOfMemory pada TaskManager.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Opsional) Aktifkan filter parsing untuk mempercepat pembacaan.
  scan.only.deserialize.captured.tables.changelog.enabled: true
route:
  - source-table: order_dw.user
    sink-table: user1
  - source-table: order_dw.order
    sink-table: order2
  - source-table: order_dw.feedback
    sink-table: feedback3
sink:
  type: upsert-kafka
  name: upsert-kafka Sink
  properties.bootstrap.servers: xxxx.alikafka.aliyuncs.com:9092
  # Parameter berikut diperlukan untuk ApsaraMQ for Kafka.
  aliyun.kafka.accessKeyId: #{ak}
  aliyun.kafka.accessKeySecret: #{sk}
  aliyun.kafka.instanceId: #{instanceId}
  aliyun.kafka.endpoint: #{endpoint}
  aliyun.kafka.regionId: #{regionId}

      2. Di pojok kanan atas, klik Deploy untuk deploy job.

      3. Di bilah navigasi kiri, pilih Operations Center > Deployments, klik Start pada kolom Actions job target, pilih Initial Mode, lalu klik Start.

      Nama topik batch

      Anda dapat menggunakan modul route untuk menentukan pola nama topik yang dihasilkan. Job berikut membuat tiga topik: topic_user, topic_order, dan topic_feedback.

      1. Pada halaman Development > Data Ingestion, buat job data ingestion Flink CDC dan salin kode berikut ke editor YAML.

        source:
  type: mysql
  name: MySQL Source
  hostname: #{hostname}
  port: 3306
  username: #{usernmae}
  password: #{password}
  tables: order_dw.\.*
  server-id: 28601-28604
  # (Opsional) Sinkronkan data dari tabel yang baru dibuat selama fase inkremental.
  scan.binlog.newly-added-table.enabled: true
  # (Opsional) Sinkronkan komentar tabel dan field.
  include-comments.enabled: true
  # (Opsional) Utamakan pengiriman unbounded splits untuk menghindari potensi masalah OutOfMemory pada TaskManager.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Opsional) Aktifkan filter parsing untuk mempercepat pembacaan.
  scan.only.deserialize.captured.tables.changelog.enabled: true
route:
  - source-table: order_dw.\.*
    sink-table: topic_<>
    replace-symbol: <>
sink:
  type: upsert-kafka
  name: upsert-kafka Sink
  properties.bootstrap.servers: xxxx.alikafka.aliyuncs.com:9092
  # Parameter berikut diperlukan untuk ApsaraMQ for Kafka.
  aliyun.kafka.accessKeyId: #{ak}
  aliyun.kafka.accessKeySecret: #{sk}
  aliyun.kafka.instanceId: #{instanceId}
  aliyun.kafka.endpoint: #{endpoint}
  aliyun.kafka.regionId: #{regionId}

      2. Di pojok kanan atas, klik Deploy untuk deploy job.

      3. Di bilah navigasi kiri, klik Operations Center > Deployments. Klik Start pada kolom Actions job target, pilih Initial Mode, lalu klik Start.

    1. Konsumsi data Kafka secara real-time.

      Job data ingestion menulis data dari database MySQL hulu ke Kafka dalam format JSON. Beberapa job hilir kemudian dapat mengonsumsi data dari satu topik untuk mengambil status terbaru tabel database. Anda dapat mengonsumsi data dari tabel yang disinkronkan ke Kafka dengan salah satu cara berikut:

      Berdasarkan katalog

      Baca data dari topik Kafka dengan menggunakannya sebagai tabel sumber.

      1. Pada halaman Development > ETL, buat job SQL streaming dan salin kode berikut ke editor SQL.

        CREATE TEMPORARY TABLE print_user_proudct(
  order_id BIGINT,
  product STRING,
  user_name STRING
) WITH (
  'connector'='print',
  'logger'='true'
);
CREATE TEMPORARY TABLE print_user_feedback(
  feedback_id BIGINT,
  `comment` STRING,
  user_name STRING
) WITH (
  'connector'='print',
  'logger'='true'
);
BEGIN STATEMENT SET;      -- Diperlukan saat menulis ke beberapa sink.
-- Join informasi pesanan dengan tabel user dalam Kafka JSON Catalog untuk menampilkan username dan nama produk untuk setiap pesanan.
INSERT INTO print_user_proudct
SELECT `order`.key_id as order_id, value_product as product, `user`.value_name as user_name
FROM `kafka-catalog`.`kafka`.`order`/*+OPTIONS('properties.group.id'='<yourGroupName>', 'scan.startup.mode'='earliest-offset')*/  as `order` -- Tentukan group dan startup mode.
LEFT JOIN `kafka-catalog`.`kafka`.`user`/*+OPTIONS('properties.group.id'='<yourGroupName>', 'scan.startup.mode'='earliest-offset')*/ as `user` -- Tentukan group dan startup mode.
ON `order`.value_user_id = `user`.key_id;
-- Join ulasan dengan tabel user untuk menampilkan isi setiap ulasan dan username yang sesuai.
INSERT INTO print_user_feedback
SELECT feedback.key_id as feedback_id, value_comment as `comment`, `user`.value_name as user_name
FROM `kafka-catalog`.`kafka`.feedback/*+OPTIONS('properties.group.id'='<yourGroupName>', 'scan.startup.mode'='earliest-offset')*/  as feedback  -- Tentukan group dan startup mode.
LEFT JOIN `kafka-catalog`.`kafka`.`user`/*+OPTIONS('properties.group.id'='<yourGroupName>', 'scan.startup.mode'='earliest-offset')*/ as `user` -- Tentukan group dan startup mode.
ON feedback.value_user_id = `user`.key_id;
END;      -- Diperlukan saat menulis ke beberapa sink.

        Contoh ini menggunakan konektor Print untuk mencetak hasil secara langsung. Anda juga dapat mengeluarkan hasil ke tabel hasil yang menggunakan konektor lain untuk analisis lebih lanjut. Untuk informasi selengkapnya tentang sintaksis penulisan ke beberapa sink, lihat Pernyataan INSERT INTO.

        Catatan

        Saat menggunakan metode ini secara langsung, perubahan skema mungkin terjadi. Akibatnya, skema yang diurai oleh Kafka JSON catalog dapat berbeda dari skema tabel MySQL yang sesuai. Misalnya, field yang dihapus mungkin masih muncul, dan beberapa field mungkin memiliki nilai null.

        Skema yang dibaca dari catalog terdiri dari field-field dari data yang dikonsumsi. Jika sebuah field dihapus tetapi pesannya belum kedaluwarsa, field tersebut mungkin masih muncul dengan nilai null. Tidak diperlukan penanganan khusus untuk kasus ini.

      2. Di pojok kanan atas, klik Deploy untuk deploy job.

      3. Di bilah navigasi kiri, klik O&M Center > Deployments, klik Start pada kolom Actions job target, pilih Initial Mode, lalu klik Start.

      Menggunakan tabel sementara

      Tentukan skema kustom dan baca data dari tabel sementara.

      1. Pada halaman Development > ETL, buat job SQL streaming dan salin kode berikut ke editor SQL.

        CREATE TEMPORARY TABLE user_source (
  key_id BIGINT,
  value_name STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'user',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'value_',
  'value.fields-include' = 'EXCEPT_KEY',
  'value.json.infer-schema.flatten-nested-columns.enable' = 'false',
  'value.json.infer-schema.primitive-as-string' = 'false'
);
CREATE TEMPORARY TABLE order_source (
  key_id  BIGINT,
  value_product STRING,
  value_user_id BIGINT  
) WITH (
  'connector' = 'kafka',
  'topic' = 'order',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'value_',
  'value.fields-include' = 'EXCEPT_KEY',
  'value.json.infer-schema.flatten-nested-columns.enable' = 'false',
  'value.json.infer-schema.primitive-as-string' = 'false'
);
CREATE TEMPORARY TABLE feedback_source (
  key_id  BIGINT,
  value_user_id BIGINT,
  value_comment STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'feedback',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'value_',
  'value.fields-include' = 'EXCEPT_KEY',
  'value.json.infer-schema.flatten-nested-columns.enable' = 'false',
  'value.json.infer-schema.primitive-as-string' = 'false'
);
CREATE TEMPORARY TABLE print_user_proudct(
  order_id BIGINT,
  product STRING,
  user_name STRING
) WITH (
  'connector'='print',
  'logger'='true'
);
CREATE TEMPORARY TABLE print_user_feedback(
  feedback_id BIGINT,
  `comment` STRING,
  user_name STRING
) WITH (
  'connector'='print',
  'logger'='true'
);
BEGIN STATEMENT SET;      -- Diperlukan saat menulis ke beberapa sink.
-- Join informasi pesanan dengan tabel user dari Kafka JSON Catalog untuk menampilkan username dan nama produk untuk setiap pesanan.
INSERT INTO print_user_proudct
SELECT order_source.key_id as order_id, value_product as product, user_source.value_name as user_name
FROM order_source LEFT JOIN user_source
ON order_source.value_user_id = user_source.key_id;
-- Join ulasan dengan tabel user untuk menampilkan isi setiap ulasan dan username yang sesuai.
INSERT INTO print_user_feedback
SELECT feedback_source.key_id as feedback_id, value_comment as `comment`, user_source.value_name as user_name
FROM feedback_source  LEFT JOIN user_source
ON feedback_source.value_user_id = user_source.key_id;
END;      -- Diperlukan saat menulis ke beberapa sink.

        Contoh ini menggunakan konektor Print untuk mencetak hasil secara langsung. Anda juga dapat mengeluarkan hasil ke tabel hasil dari konektor lain untuk analisis lebih lanjut. Untuk informasi selengkapnya tentang sintaksis penulisan ke beberapa sink, lihat Pernyataan INSERT INTO.

        Tabel berikut menjelaskan parameter konfigurasi untuk tabel sementara.

        Parameter

        Deskripsi

        Catatan

        connector

        Jenis konektor.

        Atur nilainya ke kafka.

        topic

        Nama topik yang sesuai.

        Harus konsisten dengan deskripsi Kafka JSON catalog.

        properties.bootstrap.servers

        Alamat broker Kafka.

        Formatnya adalah host:port,host:port,host:port, dipisahkan dengan koma (,).

        scan.startup.mode

        Posisi startup untuk membaca data dari Kafka.

        Nilai yang valid:

        • earliest-offset: Mulai membaca dari offset paling awal yang tersedia.

        • latest-offset: Mulai membaca dari offset terbaru.

        • group-offsets (default): Membaca dari offset yang telah dikomit untuk grup yang ditentukan oleh properties.group.id.

        • timestamp: Membaca dari timestamp yang ditentukan oleh scan.startup.timestamp-millis.

        • specific-offsets: Mulai membaca dari offset yang ditentukan dalam scan.startup.specific-offsets.

        Catatan

        Parameter ini berlaku saat job dimulai tanpa state yang disimpan. Saat job dimulai ulang atau pulih dari checkpoint, job akan memprioritaskan pembacaan dari state yang disimpan.

        key.format

        Format yang digunakan konektor Flink Kafka untuk serialisasi atau deserialisasi kunci pesan Kafka.

        Atur nilainya ke json.

        key.fields

        Field-field dalam tabel sumber atau hasil yang sesuai dengan kunci pesan Kafka.

        Gunakan titik koma (;) untuk memisahkan beberapa nama field. Misalnya, field1;field2.

        key.fields-prefix

        Awalan kustom untuk semua field kunci pesan Kafka untuk menghindari konflik penamaan dengan field nilai pesan atau field metadata.

        Nilai ini harus konsisten dengan nilai parameter key.fields-prefix dari Kafka JSON Catalog.

        value.format

        Format yang digunakan konektor Flink Kafka untuk serialisasi atau deserialisasi nilai pesan Kafka.

        Atur nilainya ke json.

        value.fields-prefix

        Awalan kustom untuk semua field nilai pesan Kafka untuk menghindari konflik penamaan dengan field kunci pesan atau field metadata.

        Harus sesuai dengan nilai parameter value.fields-prefix dari Kafka JSON Catalog.

        value.fields-include

        Kebijakan untuk menangani field kunci pesan dalam nilai pesan.

        Atur nilainya ke EXCEPT_KEY. Ini menunjukkan bahwa nilai pesan tidak mencakup field-field dari kunci pesan.

        value.json.infer-schema.flatten-nested-columns.enable

        Menentukan apakah akan memperluas secara rekursif kolom JSON bersarang dalam nilai pesan Kafka.

        Nilai parameter infer-schema.flatten-nested-columns.enable dari Catalog yang sesuai.

        value.json.infer-schema.primitive-as-string

        Menentukan apakah akan menginferensi semua tipe primitif sebagai String dalam nilai pesan Kafka.

        Nilai parameter infer-schema.primitive-as-string dari Catalog yang sesuai.

      2. Di pojok kanan atas, klik Deploy untuk deploy job.

      3. Di bilah navigasi kiri, klik Operations Center > Deployments, klik Start pada kolom Actions job target, pilih Initial Mode, lalu klik Start.

    2. Lihat hasil job.

      1. Di bilah navigasi kiri, klik Operations Center > Deployments, lalu klik job target.

      2. Pada tab Job Log, pada tab Running TaskManagers, klik task yang memiliki Path, ID yang ingin Anda lihat.

      3. Klik Logs dan cari informasi log terkait PrintSinkOutputWriter.

        Cari output PrintSinkOutputWriter dalam log. Output berisi empat catatan data hasil join: +I[1, Good., Tom], +I[2, Very good, Jerry], +I[2, Basket, Tom], dan +I[1, Football, Jerry]. Hal ini menunjukkan bahwa join antara tabel user dengan tabel order dan feedback berhasil.

    Dokumen Terkait