Stream data MySQL ke tabel Hudi melalui Flink CDC & DLF-Data Lake Formation-Alibaba Cloud

Seiring pertumbuhan database OLTP, tim sering membagi data ke beberapa database dan tabel demi meningkatkan throughput—namun hal ini menyulitkan analisis terpadu. Data Lake Formation (DLF) terintegrasi dengan Realtime Compute for Apache Flink (berbasis Ververica Platform/VVP) dan Flink Change Data Capture (CDC) untuk mengonsolidasikan data tersebut ke dalam data lake secara real time. DLF kemudian menyediakan manajemen metadata terpusat, sehingga Anda dapat menjalankan kueri terhadap data yang sama dari berbagai mesin analitik—seperti E-MapReduce (EMR), MaxCompute, dan Hologres—tanpa perlu memindahkannya lagi.

Tutorial ini memandu Anda melalui pipeline end-to-end: menangkap perubahan dari database MySQL menggunakan Flink CDC, menulisnya ke tabel hasil Apache Hudi, menyinkronkan metadata tabel ke katalog DLF, dan menjalankan kueri terhadap data lake menggunakan Flink SQL—semuanya tanpa menulis kode Java atau Scala.

Cara kerja

Architecture diagram showing MySQL → Flink CDC → Apache Hudi result table → DLF catalog → analytics engines

Realtime Compute for Apache Flink membaca event perubahan dari sumber MySQL menggunakan konektor mysql-cdc. Flink menuliskan perubahan tersebut ke tabel hasil Apache Hudi sekaligus menyinkronkan skema tabel ke katalog DLF. Setelah metadata terdaftar di DLF, Anda dapat menjalankan kueri langsung terhadap tabel Hudi dari konsol Flink SQL atau dari mesin komputasi yang terhubung. DLF juga menangani manajemen siklus hidup data lake dan optimasi format lake, sehingga data tetap dapat diakses dan hemat biaya seiring waktu.

Prasyarat

Sebelum memulai, pastikan Anda telah:

Realtime Compute for Apache Flink diaktifkan, dengan ruang kerja Flink yang sepenuhnya dikelola telah dibuat. Lihat Memulai Realtime Compute for Apache Flink.
DLF diaktifkan. Jika belum diaktifkan, buka halaman produk Data Lake Formation dan klik Activate Now.
Instans ApsaraDB RDS for MySQL di wilayah dan VPC yang sama dengan ruang kerja Realtime Compute for Apache Flink, menjalankan MySQL 5.7 atau versi lebih baru. Untuk instruksi penyiapan, lihat Buat instans ApsaraDB RDS for MySQL. Lewati langkah ini jika Anda menggunakan database sumber lain.

Langkah 1: Siapkan data MySQL

Jalankan SQL berikut untuk membuat database dan tabel uji, lalu masukkan data sampel.

CREATE DATABASE testdb;
CREATE TABLE testdb.student (
  `id` bigint(20) NOT NULL,
  `name` varchar(256) DEFAULT NULL,
  `age` bigint(20) DEFAULT NULL,
  PRIMARY KEY (`id`)
);

INSERT INTO testdb.student VALUES (1,'name1',10);
INSERT INTO testdb.student VALUES (2,'name2',20);

Langkah 2: Buat katalog DLF di Flink

Login ke konsol Realtime Compute for Apache Flink.
Pada tab Fully Managed Flink, klik Console di kolom Actions ruang kerja target.
Di panel navigasi kiri, klik Catalogs, lalu klik Create Catalog.
Pada halaman Create Catalog, pilih DLF dan klik Next.
Masukkan konfigurasi katalog, lalu klik Confirm. Untuk detail parameter, lihat Kelola katalog DLF.

Setelah katalog dibuat, katalog tersebut muncul sebagai dlf di bawah Catalogs. Ini adalah katalog data default untuk DLF.

Langkah 3: Buat pekerjaan data lake Flink

Buat tabel sumber dan tabel hasil

Di panel navigasi kiri, klik Development > Scripts.

Di area pengeditan SQL, masukkan SQL berikut dan klik Run.

-- Buat tabel sumber yang membaca event perubahan dari MySQL
CREATE TABLE IF NOT EXISTS student_source (
  id INT,
  name VARCHAR(256),
  age INT,
  PRIMARY KEY (id) NOT ENFORCED
)
WITH (
  'connector' = 'mysql-cdc',
  -- Ganti dengan endpoint instans ApsaraDB RDS for MySQL Anda
  'hostname' = 'rm-xxxxxxxx.mysql.rds.aliyuncs.com',
  'port' = '3306',
  'username' = '<RDS username>',
  'password' = '<RDS password>',
  'database-name' = '<RDS database>',
  -- Atur ke nama tabel sumber yang dibuat di Langkah 1
  'table-name' = 'student'
);

-- Buat database target di katalog DLF
-- Ganti 'dlf' dengan nama katalog DLF Anda jika berbeda
CREATE DATABASE IF NOT EXISTS dlf.dlf_testdb;

-- Buat tabel hasil Apache Hudi di katalog DLF
CREATE TABLE IF NOT EXISTS dlf.dlf_testdb.student_hudi (
  id    BIGINT PRIMARY KEY NOT ENFORCED,
  name  STRING,
  age   BIGINT
) WITH (
  'connector' = 'hudi'
);

Placeholder	Deskripsi	Contoh
`rm-xxxxxxxx.mysql.rds.aliyuncs.com`	Endpoint instans ApsaraDB RDS for MySQL	`rm-bp1xxxxxxxx.mysql.rds.aliyuncs.com`
`<RDS username>`	Pengguna MySQL dengan akses baca ke tabel sumber	`admin`
`<RDS password>`	Kata sandi untuk pengguna MySQL	—
`<RDS database>`	Nama database sumber	`testdb`

Setelah tabel dibuat, keduanya muncul di bawah Catalogs.

Catalogs pane showing the newly created source and result tables

Untuk daftar lengkap parameter konektor mysql-cdc, lihat Konektor sumber MySQL. Untuk parameter tabel hasil Hudi, lihat Konektor Hudi (akan dipensiunkan).

Buat dan sebarkan pekerjaan streaming

Di panel navigasi kiri, klik Development > ETL.
Klik New, pilih Blank Stream Draft di kotak dialog New Draft, lalu klik Next.
Masukkan konfigurasi draft dan klik Create.

Di area pengeditan SQL, masukkan pernyataan INSERT berikut.

-- Streaming perubahan dari sumber MySQL ke tabel hasil Hudi
INSERT INTO dlf.dlf_testdb.student_hudi
SELECT * FROM student_source /*+ OPTIONS('server-id'='123456') */;

Di pojok kanan atas area pengeditan SQL, klik Deploy. Di kotak dialog Deploy draft, isi kolom yang diperlukan lalu klik Confirm.

Jalankan pekerjaan

Di panel navigasi kiri, klik O&M > Deployments.
Di kolom Actions pekerjaan target, klik Start.
Pilih Initial Mode lalu klik Start.

Saat status pekerjaan berubah menjadi RUNNING, Flink secara aktif menangkap perubahan dari MySQL dan menuliskannya ke tabel hasil Hudi. Untuk detail parameter startup, lihat Jalankan deployment.

Langkah 4: Verifikasi data lake

Di panel navigasi kiri, klik Development > Scripts.
Jalankan kueri berikut untuk memastikan baris awal telah ditulis ke data lake.
```
SELECT * FROM dlf.dlf_testdb.student_hudi;
```
Hasil menampilkan dua baris yang dimasukkan di Langkah 1.

Jika Anda memiliki kluster EMR dengan metadata DLF yang diaktifkan, Anda juga dapat menjalankan kueri terhadap tabel Hudi melalui kluster EMR. Lihat Integrasikan Hudi dengan Spark SQL.

Langkah selanjutnya

Untuk membaca dari dan menulis ke tabel Hudi di DLF menggunakan Flink di kluster EMR DataFlow, lihat Gunakan kluster Dataflow untuk membaca dan menulis data ke tabel Hudi berbasis DLF.