Memulai cepat pekerjaan ingesti data Flink CDC - Realtime Compute for Apache Flink

Realtime Compute for Apache Flink menyediakan fitur ingesti data andal berbasis Flink CDC. Panduan ini menjelaskan cara membuat pekerjaan ingesti data Flink CDC untuk menyinkronkan seluruh database MySQL ke database StarRocks.

Prasyarat

Ruang kerja Flink telah dibuat. Untuk informasi lebih lanjut, lihat Aktifkan Realtime Compute for Apache Flink.
Penyimpanan data sumber dan sink
- Instans ApsaraDB RDS for MySQL telah dibuat. Untuk informasi lebih lanjut, lihat (Usang, dialihkan ke "Langkah 1") Buat instans ApsaraDB RDS for MySQL dengan cepat.
- Instans EMR Serverless StarRocks telah dibuat. Untuk informasi lebih lanjut, lihat Prosedur.
Catatan
Instans ApsaraDB RDS for MySQL dan instans EMR Serverless StarRocks harus berada dalam Virtual Private Cloud (VPC) yang sama dengan ruang kerja Flink Anda. Jika berada di VPC yang berbeda, Anda harus membuat koneksi jaringan dan mengonfigurasi daftar izin (allowlist) untuk instans ApsaraDB RDS for MySQL. Untuk informasi lebih lanjut, lihat Bagaimana cara mengakses layanan lain lintas VPC?, Bagaimana cara mengakses Internet?, dan Bagaimana cara mengonfigurasi daftar izin?.

Informasi latar belakang

Asumsikan bahwa instans ApsaraDB RDS for MySQL Anda memiliki database bernama order_dw_mysql yang berisi tiga tabel bisnis: orders, orders_pay, dan product_catalog. Untuk menyinkronkan tabel-tabel ini beserta datanya ke database order_dw_sr di StarRocks, ikuti langkah-langkah berikut:

Langkah 1: Siapkan data uji di ApsaraDB RDS for MySQL
Langkah 2: Kembangkan pekerjaan ingesti data Flink CDC
Langkah 3: Jalankan pekerjaan ingesti data Flink CDC
Langkah 4: Verifikasi hasil sinkronisasi di StarRocks

Langkah 1: Siapkan data uji MySQL

Buat database dan akun.

Buat database bernama order_dw_mysql dan akun standar dengan izin baca dan tulis ke database tersebut. Untuk informasi lebih lanjut, lihat (Usang, dialihkan ke "Langkah 1") Buat database dan akun dan Kelola database.
Login ke instans ApsaraDB RDS for MySQL menggunakan Data Management (DMS).

Untuk informasi lebih lanjut, lihat (Usang, dialihkan ke "Langkah 2") Login ke instans ApsaraDB RDS for MySQL menggunakan DMS.

Di jendela SQL Console, masukkan perintah berikut dan klik Execute untuk membuat tiga tabel bisnis dan memasukkan data.

CREATE TABLE `orders` (
  order_id bigint not null primary key,
  user_id varchar(50) not null,
  shop_id bigint not null,
  product_id bigint not null,
  buy_fee numeric(20,2) not null,   
  create_time timestamp not null,
  update_time timestamp not null default now(),
  state int not null 
);
CREATE TABLE `orders_pay` (
  pay_id bigint not null primary key,
  order_id bigint not null,
  pay_platform int not null, 
  create_time timestamp not null
);
CREATE TABLE `product_catalog` (
  product_id bigint not null primary key,
  catalog_name varchar(50) not null
);
-- Siapkan data
INSERT INTO product_catalog VALUES(1, 'phone_aaa'),(2, 'phone_bbb'),(3, 'phone_ccc'),(4, 'phone_ddd'),(5, 'phone_eee');
INSERT INTO orders VALUES
(100001, 'user_001', 12345, 1, 5000.05, '2023-02-15 16:40:56', '2023-02-15 18:42:56', 1),
(100002, 'user_002', 12346, 2, 4000.04, '2023-02-15 15:40:56', '2023-02-15 18:42:56', 1),
(100003, 'user_003', 12347, 3, 3000.03, '2023-02-15 14:40:56', '2023-02-15 18:42:56', 1),
(100004, 'user_001', 12347, 4, 2000.02, '2023-02-15 13:40:56', '2023-02-15 18:42:56', 1),
(100005, 'user_002', 12348, 5, 1000.01, '2023-02-15 12:40:56', '2023-02-15 18:42:56', 1),
(100006, 'user_001', 12348, 1, 1000.01, '2023-02-15 11:40:56', '2023-02-15 18:42:56', 1),
(100007, 'user_003', 12347, 4, 2000.02, '2023-02-15 10:40:56', '2023-02-15 18:42:56', 1);
INSERT INTO orders_pay VALUES
(2001, 100001, 1, '2023-02-15 17:40:56'),
(2002, 100002, 1, '2023-02-15 17:40:56'),
(2003, 100003, 0, '2023-02-15 17:40:56'),
(2004, 100004, 0, '2023-02-15 17:40:56'),
(2005, 100005, 0, '2023-02-15 18:40:56'),
(2006, 100006, 0, '2023-02-15 18:40:56'),
(2007, 100007, 0, '2023-02-15 18:40:56');

Langkah 2: Kembangkan pekerjaan Flink CDC

Login ke Konsol manajemen Realtime Compute for Apache Flink.
Klik Console untuk membuka ruang kerja proyek.
Di panel navigasi kiri, pilih Development > Data Ingestion.
Klik ikon , klik New Draft with Template, pilih MySQL to StarRocks data synchronization, lalu klik Next.
Masukkan Job Name dan Storage Location, pilih Engine Version, lalu klik OK.

Konfigurasikan kode pekerjaan YAML.

Kode berikut memberikan contoh untuk menyinkronkan semua tabel dari database order_dw_mysql di MySQL ke database order_dw_sr di StarRocks.

source:
  type: mysql
  hostname: rm-bp1rk934iidc3****.mysql.rds.aliyuncs.com
  port: 3306
  username: ${secret_values.mysqlusername}
  password: ${secret_values.mysqlpassword}
  tables: order_dw_mysql.\.*
  server-id: 8601-8604
  # (Opsional) Sinkronkan data dari tabel yang baru dibuat selama fase inkremental.
  scan.binlog.newly-added-table.enabled: true
  # (Opsional) Sinkronkan komentar tabel dan kolom.
  include-comments.enabled: true
  # (Opsional) Utamakan distribusi chunk tak terbatas untuk mencegah potensi error 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: starrocks
  name: StarRocks Sink
  jdbc-url: jdbc:mysql://fe-c-b76b6aa51807****-internal.starrocks.aliyuncs.com:9030
  load-url: fe-c-b76b6aa51807****-internal.starrocks.aliyuncs.com:8030
  username: ${secret_values.starrocksusername}
  password: ${secret_values.starrockspassword}
  table.create.properties.replication_num: 1
  sink.buffer-flush.interval-ms: 5000 # Flush data setiap 5 detik.
route:
  - source-table: order_dw_mysql.\.*
    sink-table: order_dw_sr.<>
    replace-symbol: <>
    description: route all tables in source_db to sink_db
pipeline:
  name: Sync MySQL Database to StarRocks

Tabel berikut menjelaskan parameter konfigurasi yang diperlukan untuk contoh ini. Untuk informasi lebih lanjut tentang parameter ingesti data, lihat MySQL dan StarRocks.

Catatan

Pekerjaan YAML hanya mendukung variabel proyek. Anda dapat menggunakan variabel untuk mencegah informasi seperti password ditampilkan dalam teks biasa. Untuk informasi lebih lanjut, lihat Manajemen Variabel.

Kategori	Parameter	Deskripsi	Nilai contoh
source	hostname	Alamat IP atau hostname database MySQL. Kami menyarankan menggunakan titik akhir internal.	`rm-bp1rk934iidc3****.mysql.rds.aliyuncs.com`
	port	Nomor port layanan database MySQL.	3306
	username	Username dan password untuk database MySQL. Gunakan kredensial akun yang dibuat di Langkah 1: Siapkan data uji di ApsaraDB RDS for MySQL.	`${secret_values.mysqlusername}`
	password		`${secret_values.mysqlpassword}`
	tables	Nama tabel MySQL. Anda dapat menggunakan ekspresi reguler untuk membaca data dari beberapa tabel. Dalam topik ini, semua tabel dan data di database order_dw_mysql disinkronkan.	order_dw_mysql.\.*
	server-id	ID numerik unik untuk koneksi client database.	8601-8604
sink	jdbc-url	URL koneksi JDBC. Tentukan alamat IP dan port kueri Frontend (FE) dalam format `jdbc:mysql://ip:port`. Di tab Instance Details di Konsol E-MapReduce, Anda dapat melihat internal endpoint dan query port FE untuk instans target.	`jdbc:mysql://fe-c-b76b6aa51807****-internal.starrocks.aliyuncs.com:9030`
	load-url	URL layanan HTTP yang digunakan untuk menghubungkan ke node FE. Di tab Instance Details di Konsol E-MapReduce, Anda dapat melihat internal endpoint dan HTTP port FE untuk instans target.	`fe-c-b76b6aa51807****-internal.starrocks.aliyuncs.com:8030`
	username	Kredensial untuk menghubungkan ke StarRocks. Gunakan kredensial yang Anda konfigurasikan saat membuat instans StarRocks. Catatan Contoh ini menggunakan variabel untuk menghindari pengeksposan kredensial dalam teks biasa. Untuk informasi lebih lanjut, lihat Kelola variabel.	`${secret_values.starrocksusername}`
	password		`${secret_values.starrockspassword}`
	sink.buffer-flush.interval-ms	Interval flush untuk buffer internal. Interval pendek (5 detik) digunakan karena contoh ini berisi jumlah data yang kecil, sehingga Anda dapat melihat hasilnya dengan cepat.	5000
route	source-table	Tabel sumber yang akan dirutekan. Anda dapat menggunakan ekspresi reguler untuk mencocokkan beberapa tabel. Misalnya, `order_dw_mysql.\.*` merutekan semua tabel di database `order_dw_mysql`.	order_dw_mysql.\.*
	sink-table	Pola tabel tujuan untuk data yang dirutekan. Anda dapat menggunakan simbol yang didefinisikan dalam parameter `replace-symbol` sebagai placeholder untuk setiap nama tabel sumber guna mencapai routing banyak-ke-banyak. Untuk informasi lebih lanjut tentang aturan routing, lihat Modul route.	order_dw_sr.<>
	replace-symbol	Placeholder untuk nama tabel sumber yang digunakan dalam pencocokan pola.	<>

Klik Deploy.

Langkah 3: Jalankan pekerjaan Flink CDC

Di halaman Data Ingestion, klik Deploy lalu klik OK di kotak dialog yang muncul.
Di halaman O&M > Deployments, temukan pekerjaan YAML target dan klik Start di kolom Actions.
Klik Start.

Dalam contoh ini, pilih Initial Mode. Untuk informasi lebih lanjut tentang parameter, lihat Jalankan pekerjaan. Setelah pekerjaan dimulai, Anda dapat memantau statusnya di halaman Deployments.

Langkah 4: Verifikasi hasil di StarRocks

Setelah pekerjaan memasuki status RUNNING, Anda dapat memverifikasi data di StarRocks.

Hubungkan ke instans EMR Serverless StarRocks menggunakan EMR StarRocks Manager.
Di panel navigasi kiri, klik SQL Editor. Di tab Database, klik ikon refresh .

Database bernama order_dw_sr muncul di bawah default_catalog.

Di tab Query List, klik + File untuk membuat Query Script. Masukkan pernyataan SQL berikut dan klik Run.

SELECT * FROM default_catalog.order_dw_sr.orders order by order_id;
SELECT * FROM default_catalog.order_dw_sr.orders_pay order by pay_id;
SELECT * FROM default_catalog.order_dw_sr.product_catalog order by product_id;

Lihat hasil di bawah perintah.

Hasil menunjukkan bahwa tabel dan data dari database MySQL kini tersedia di StarRocks.

Tabel yang disinkronkan mencakup default_catalog.order_dw_sr.orders, default_catalog.order_dw_sr.orders_pay, dan default_catalog.order_dw_sr.product_catalog. Anda dapat menjalankan pernyataan SELECT untuk mengkueri setiap tabel dan memverifikasi integritas data.

Dokumentasi terkait

Untuk langkah-langkah detail mengenai pengembangan pekerjaan ingesti data Flink CDC, lihat Kembangkan pekerjaan ingesti data Flink CDC.
Untuk informasi lebih lanjut tentang modul source, sink, transform, dan route untuk pekerjaan ingesti data Flink CDC, lihat Referensi pengembangan pekerjaan ingesti data Flink CDC.