MaxCompute：Flink CDC Open-source untuk Penulisan Waktu-Nyata Mendekati ke Tabel Delta

Siapkan Lingkungan

Kirim File YAML Menggunakan CLI Flink CDC

1. Unduh paket JAR yang diperlukan:

   • paket flink-cdc

     Pergi ke halaman flink-cdc untuk mengunduh paket biner terkompresi flink-cdc-3.1.1-bin.tar.gz dan dekompresi paket tersebut untuk mendapatkan direktori flink-cdc-3.1.1. Direktori flink-cdc-3.1.1 berisi direktori bin, lib, log, dan conf. Kemudian, pindahkan file-file dalam direktori-direktori ini ke direktori terkait untuk flink-1.18.0.

   • Paket konektor

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

     Catatan

     Anda dapat mengklik tautan unduhan untuk mengunduh hanya versi konektor yang dirilis. Jika Anda ingin merujuk versi SNAPSHOT, Anda harus mengkompilasi kode sumber versi tersebut berdasarkan cabang master atau rilis di mesin lokal Anda.

     • Konektor Pipeline MySQL 3.1.1

     • Konektor Pipeline MaxCompute 3.1.1

   • Paket driver

     Unduh paket MySQL Connector Java dan teruskan paket tersebut ke CLI Flink CDC menggunakan parameter --jar, atau letakkan paket tersebut di direktori $flink-1.18.0/lib dan mulai ulang kluster Flink. Ini karena konektor CDC tidak lagi berisi driver-driver ini.

2. Tulis file YAML untuk konfigurasi tugas. Contoh kode berikut memberikan contoh file mysql-to-maxcompute.yaml untuk sinkronisasi database:

   ################################################################################
   # 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
   
   # Konfigurasikan parameter accessId, accessKey, endpoint, dan project.
   sink:
      type: maxcompute
      name: MaxComputeSink
      accessId: ${your_accessId}
      accessKey: ${your_accessKey}
      endpoint: ${your_maxcompute_endpoint}
      project: ${your_project}
      bucketsNum: 8
   
   pipeline:
     name: Sinkronkan Database MySQL ke MaxCompute
     parallelism: 1
   

   Parameter:

   • Untuk informasi lebih lanjut tentang parameter di bagian sumber, lihat Konektor MySQL.

   • Untuk informasi lebih lanjut tentang parameter di bagian sink, lihat Item Konfigurasi Konektor.

3. Jalankan perintah berikut untuk mengirimkan tugas ke kluster Flink yang diterapkan dalam mode mandiri:

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

   Setelah tugas dikirimkan, hasil berikut dikembalikan:

   Pipeline telah dikirimkan ke kluster.
   Job ID: f9f9689866946e25bf151ecc179ef46f
   Deskripsi Job: Sinkronkan Database MySQL ke MaxCompute

   Di UI web Flink, tugas bernama Sinkronkan Database MySQL ke MaxCompute sedang berjalan.

4. Eksekusi pernyataan SQL berikut di MaxCompute untuk memeriksa apakah tabel orders, shipments, dan products telah dibuat dan data dapat ditulis ke tabel-tabel tersebut.

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

Sinkronkan Perubahan Secara Real-Time

Dalam contoh ini, tabel orders digunakan. Saat data di tabel sumber dalam database MySQL diubah, data di tabel tujuan MaxCompute juga diubah secara real-time.

1. Jalankan perintah berikut untuk mengakses kontainer MySQL:

   docker-compose exec mysql mysql -uroot -p123456

2. Masukkan catatan data ke tabel orders MySQL.

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

   Jalankan perintah read orders; di MaxCompute untuk menanyakan data di tabel orders. Hasil berikut dikembalikan:

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

3. Tambahkan bidang ke tabel orders MySQL.

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

   Jalankan perintah read orders; di MaxCompute untuk menanyakan data di tabel orders. Hasil berikut dikembalikan:

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

4. Perbarui catatan data di tabel orders MySQL.

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

   Jalankan perintah read orders; di MaxCompute untuk menanyakan data di tabel orders. Hasil berikut dikembalikan:

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

5. Hapus catatan data dari tabel orders MySQL.

   DELETE FROM app_db.orders WHERE id=2;

   Jalankan perintah read orders; di MaxCompute untuk menanyakan data di tabel orders. Hasil berikut dikembalikan:

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

Setiap kali Anda melakukan operasi di MySQL, pratinjau data dilakukan di MaxCompute. Data yang ditampilkan di tabel orders di MaxCompute diperbarui secara real-time.

Routing Perubahan

Flink CDC menyediakan fitur untuk merutekan skema atau data tabel sumber ke nama tabel lain. Anda dapat menggunakan fitur ini untuk melakukan operasi seperti penggantian nama tabel atau database, dan sinkronisasi database. Contoh kode berikut memberikan contoh.

################################################################################
# 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

# Konfigurasikan parameter accessId, accessKey, endpoint, dan project.
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: Sinkronkan Database MySQL ke MaxCompute
   parallelism: 1

Untuk informasi lebih lanjut tentang parameter di bagian route, lihat Route.

Konfigurasi sebelumnya di bagian route digunakan untuk menyinkronkan skema dan data tabel app_db.orders ke tabel ods_db.ods_orders. Dengan cara ini, migrasi database diimplementasikan. Anda dapat menggunakan ekspresi reguler untuk mencocokkan beberapa tabel di source-table untuk menyinkronkan data dari beberapa tabel dalam database sharded. Contoh kode:

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

Dalam kasus ini, data dari tabel-tabel, seperti app_db.order01, app_db.order02, dan app_db.order03, dapat digabungkan ke dalam tabel ods_db.ods_orders.

Bersihkan Lingkungan

Setelah Anda melakukan operasi sebelumnya, Anda harus membersihkan lingkungan.

1. Jalankan perintah berikut di direktori tempat file docker-compose.yml disimpan untuk menghentikan semua kontainer:

   docker-compose down

2. Jalankan perintah berikut di direktori flink-1.18.0 tempat Flink berada untuk menghentikan kluster Flink:

   ./bin/stop-cluster.sh

Item konfigurasi konektor

Pemetaan Lokasi Tabel

Saat konektor Flink CDC membuat tabel secara otomatis, informasi lokasi tabel sumber dipetakan ke tabel MaxCompute berdasarkan pemetaan yang dijelaskan dalam tabel berikut.

Pemetaan Tipe Data