MaxCompute:Menulis ke Delta Table secara near-real-time menggunakan Flink CDC open source

Persiapan Lingkungan

Mengirimkan Tugas melalui Flink CDC CLI

1. Unduh paket JAR yang diperlukan:

   • Paket flink-cdc

     Masuk ke flink-cdc untuk mengunduh arsip biner flink-cdc-3.1.1-bin.tar.gz, lalu ekstrak untuk mendapatkan direktori flink-cdc-3.1.1 yang berisi empat subdirektori: bin, lib, log, dan conf. Pindahkan file dari keempat direktori tersebut ke direktori flink-1.18.0 yang sesuai.

   • Paket Connector

     Unduh paket Connector berikut dan pindahkan ke direktori flink-1.18.0/lib.

     Catatan

     Tautan unduh hanya berlaku untuk versi yang telah dirilis. Versi SNAPSHOT harus dikompilasi secara lokal dari branch master atau release.

     • MySQL pipeline connector 3.1.1

     • MaxCompute pipeline connector 3.1.1

   • Paket Driver

     Unduh paket MySQL Connector Java, lalu sertakan melalui parameter --jar saat menjalankan Flink CDC CLI, atau letakkan di direktori $flink-1.18.0/lib dan mulai ulang kluster Flink, karena CDC Connectors tidak lagi menyertakan driver-driver tersebut.

2. Buat file konfigurasi tugas dalam format YAML. Berikut contoh file sinkronisasi seluruh database mysql-to-maxcompute.yaml:

   ################################################################################
   # Deskripsi: Sinkronkan semua tabel MySQL ke MaxCompute
   ################################################################################
   source:
     type: mysql
     hostname: localhost
     port: 3306
     username: root
     password: 123456
     tables: app_db.\.*
     server-id: 5400-5404
     server-time-zone: UTC
   
   # accessId, accessKey, endpoint, project harus diisi oleh pengguna
   sink:
      type: maxcompute
      name: MaxComputeSink
      accessId: ${your_accessId}
      accessKey: ${your_accessKey}
      endpoint: ${your_maxcompute_endpoint}
      project: ${your_project}
      bucketsNum: 8
   
   pipeline:
     name: Sync MySQL Database to MaxCompute
     parallelism: 1
   

   Penjelasan parameter:

   • Detail konfigurasi parameter bagian Source dapat dilihat di Apache Flink CDC MySQL Connector.

   • Cara konfigurasi parameter bagian Sink dapat dilihat di Opsi Konfigurasi Connector.

3. Jalankan perintah berikut untuk mengirimkan tugas ke kluster Flink Standalone.

   ./bin/flink-cdc.sh mysql-to-maxcompute.yaml

   Jika berhasil, Anda akan menerima respons berikut:

   Pipeline has been submitted to cluster.
   Job ID: f9f9689866946e25bf151ecc179ef46f
   Job Description: Sync MySQL Database to MaxCompute

   Di Flink Web UI, Anda akan melihat tugas bernama Sync MySQL Database to MaxCompute sedang berjalan.

4. Di MaxCompute, jalankan SQL berikut untuk memverifikasi apakah ketiga tabel orders, shipments, dan products telah berhasil dibuat dan dapat menerima data.

   -- Lihat tabel orders
   read orders;
   
   -- Hasil:
   +------------+------------+
   | id         | price      |
   +------------+------------+
   | 1          | 4          |
   | 2          | 100        |
   +------------+------------+
   
   -- Lihat tabel shipments
   read shipments;
   
   -- Hasil
   +------------+------------+
   | id         | city       |
   +------------+------------+
   | 1          | beijing    |
   | 2          | xian       |
   +------------+------------+
   
   -- Lihat tabel products
   read products;
   
   -- Hasil
   +------------+------------+
   | id         | product    |
   +------------+------------+
   | 3          | Peanut     |
   | 1          | Beer       |
   | 2          | Cap        |
   +------------+------------+

Sinkronisasi Operasi Perubahan

Berikut ini contoh menggunakan tabel orders untuk menunjukkan bahwa saat data tabel sumber di database MySQL diubah, data tabel tujuan yang sesuai di MaxCompute juga akan diperbarui secara real-time.

1. Jalankan perintah berikut untuk masuk ke kontainer MySQL.

   docker-compose exec mysql mysql -uroot -p123456

2. Masukkan satu baris data ke tabel orders di MySQL.

   INSERT INTO app_db.orders (id, price) VALUES (3, 100.00);

   Di MaxCompute, jalankan perintah read orders; untuk memeriksa data tabel orders. Hasilnya sebagai berikut:

   +------------+------------+
   | id         | price      |
   +------------+------------+
   | 3          | 100        |
   | 1          | 4          |
   | 2          | 100        |
   +------------+------------+

3. Tambahkan satu kolom ke tabel orders di MySQL.

   ALTER TABLE app_db.orders ADD amount varchar(100) NULL;

   Di MaxCompute, jalankan perintah read orders; untuk memeriksa data tabel orders. Hasilnya sebagai berikut:

   +------------+------------+------------+
   | id         | price      | amount     |
   +------------+------------+------------+
   | 3          | 100        | NULL       |
   | 1          | 4          | NULL       |
   | 2          | 100        | NULL       |
   +------------+------------+------------+

4. Perbarui satu baris data di tabel orders MySQL.

   UPDATE app_db.orders SET price=100.00, amount=100.00 WHERE id=1;

   Di MaxCompute, jalankan perintah read orders; untuk memeriksa data tabel orders. Hasilnya sebagai berikut:

   +------------+------------+------------+
   | id         | price      | amount     |
   +------------+------------+------------+
   | 3          | 100        | NULL       |
   | 1          | 100        | 100.00     |
   | 2          | 100        | NULL       |
   +------------+------------+------------+

5. Hapus satu baris data dari tabel orders MySQL.

   DELETE FROM app_db.orders WHERE id=2;

   Di MaxCompute, jalankan perintah read orders; untuk memeriksa data tabel orders. Hasilnya sebagai berikut:

   +------------+------------+------------+
   | id         | price      | amount     |
   +------------+------------+------------+
   | 3          | 100        | NULL       |
   | 1          | 100        | 100.00     |
   +------------+------------+------------+

Untuk setiap langkah di atas, setiap kali Anda menjalankan operasi di MySQL, lakukan pratinjau data di MaxCompute. Anda akan melihat bahwa data tabel orders di MaxCompute diperbarui secara real-time.

Operasi Perubahan dengan Polling

Flink CDC menyediakan konfigurasi untuk merutekan struktur tabel atau data sumber ke nama tabel lain. Dengan kemampuan ini, Anda dapat mengganti nama tabel atau database, serta melakukan sinkronisasi seluruh database. Berikut contoh file konfigurasi:

################################################################################
# Deskripsi: Sinkronkan semua tabel MySQL ke MaxCompute
################################################################################
source:
   type: mysql
   hostname: localhost
   port: 3306
   username: root
   password: 123456
   tables: app_db.\.*
   server-id: 5400-5404
   server-time-zone: UTC

# accessId, accessKey, endpoint, project harus diisi oleh pengguna
sink:
   type: maxcompute
   name: MaxComputeSink
   accessId: ${your_accessId}
   accessKey: ${your_accessKey}
   endpoint: ${your_maxcompute_endpoint}
   project: ${your_project}
   bucketsNum: 8

route:
   - source-table: app_db.orders
     sink-table: ods_db.ods_orders
   - source-table: app_db.shipments
     sink-table: ods_db.ods_shipments
   - source-table: app_db.products
     sink-table: ods_db.ods_products

pipeline:
   name: Sync MySQL Database to MaxCompute
   parallelism: 1

Detail parameter bagian route dapat dilihat di Flink CDC Route.

Dengan konfigurasi route di atas, struktur dan data tabel app_db.orders akan disinkronkan ke ods_db.ods_orders, sehingga mencapai fungsi migrasi database. Secara khusus, source-table mendukung pencocokan regex untuk beberapa tabel, memungkinkan sinkronisasi multi-database dan multi-tabel. Contohnya seperti konfigurasi berikut:

route:
  - source-table: app_db.order\.*
    sink-table: ods_db.ods_orders

Dengan demikian, data dari tabel seperti app_db.order01, app_db.order02, dan app_db.order03 dapat digabungkan ke dalam ods_db.ods_orders.

Pembersihan Lingkungan

Setelah menyelesaikan operasi di atas, Anda perlu membersihkan lingkungan.

1. Di direktori tempat file docker-compose.yml berada, jalankan perintah berikut untuk menghentikan semua kontainer:

   docker-compose down

2. Di direktori Flink flink-1.18.0, jalankan perintah berikut untuk menghentikan kluster Flink:

   ./bin/stop-cluster.sh

Opsi Konfigurasi Connector

Pemetaan Lokasi Tabel

Saat Connector membuat tabel secara otomatis, hubungan pemetaan berikut digunakan untuk memetakan informasi lokasi tabel sumber ke tabel MaxCompute.

Pemetaan Tipe Data