E-MapReduce：Gunakan pernyataan CTAS dan CDAS Realtime Compute for Apache Flink untuk menyinkronkan data dari instans ApsaraDB RDS for MySQL ke kluster StarRocks

Perilaku perubahan skema

CTAS menggunakan kebijakan tetap untuk menyebarkan perubahan skema. Tabel berikut merangkum perubahan yang didukung dan tidak didukung secara otomatis.

Perubahan skema yang didukung

Perubahan skema yang tidak didukung

Jika salah satu perubahan berikut terjadi pada tabel sumber, hapus tabel tujuan di StarRocks dan mulai ulang pekerjaan CTAS. Flink akan membuat ulang tabel tersebut dan menyinkronkan kembali semua data historis.

• Mengubah tipe data kolom (misalnya, VARCHAR menjadi BIGINT, atau NOT NULL menjadi NULLABLE)

• Mengubah constraint seperti kunci primer atau indeks

• Menambah atau menghapus kolom non-nullable

• Menyesuaikan panjang field dalam pernyataan DDL (Data Definition Language)

CTAS mendeteksi perubahan skema dengan membandingkan skema dari catatan data berturut-turut — bukan dengan mengurai jenis pernyataan DDL. Akibatnya:

Jika sebuah kolom dihapus lalu ditambahkan kembali tanpa perubahan data di antaranya, CTAS menganggap hal ini sebagai tidak ada perubahan skema.

Perubahan skema hanya disebarkan ketika data baru tiba di tabel sumber setelah perubahan tersebut.

Untuk pemetaan tipe field, lihat Continuously load data from Apache Flink.

Saat CTAS menggabungkan beberapa tabel MySQL, Flink secara otomatis menambahkan kolom _db_name dan _table_name ke tabel tujuan untuk melacak sumbernya. Tentukan urutan kolom Anda sendiri mulai dari kolom ketiga.

Semantik at-least-once (direkomendasikan untuk skenario latensi rendah)

Data ditulis pada interval flush yang dapat dikonfigurasi. Penggunaan memori lebih rendah, tetapi penulisan duplikat mungkin terjadi saat terjadi kegagalan.

/* Semantik at-least-once */

USE CATALOG sr;

CREATE TABLE IF NOT EXISTS runoob_tbl_sr WITH (
  'starrocks.create.table.properties' = 'engine = olap primary key(runoob_id) distributed by hash(runoob_id) buckets 8',
  'database-name'                     = 'test_cdc',
  'jdbc-url'                          = 'jdbc:mysql://172.16.**.**:9030',
  'load-url'                          = '172.16.**.**:18030',
  'table-name'                        = 'runoob_tbl_sr',
  'username'                          = 'test',
  'password'                          = '123456',
  'sink.buffer-flush.interval-ms'     = '5000',
  'sink.properties.row_delimiter'     = '\x02',
  'sink.properties.column_separator'  = '\x01'
)
AS TABLE mysql.test_cdc.runoob_tbl
/*+ OPTIONS (
  'connector'     = 'mysql-cdc',
  'hostname'      = 'rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com',
  'port'          = '3306',
  'username'      = 'test',
  'password'      = '123456',
  'database-name' = 'test_cdc',
  'table-name'    = 'runoob_tbl'
) */;

Semantik exactly-once (direkomendasikan untuk skenario kritis data)

Tidak ada kehilangan data atau duplikasi saat terjadi kegagalan. Visibilitas data bergantung pada interval checkpoint.

/* Semantik exactly-once */

SET 'execution.checkpointing.interval' = '1 min';
SET 'execution.checkpointing.mode'     = 'EXACTLY_ONCE';
SET 'execution.checkpointing.timeout'  = '10 min';

USE CATALOG sr;

CREATE TABLE IF NOT EXISTS runoob_tbl WITH (
  'starrocks.create.table.properties' = 'engine = olap primary key(runoob_id) distributed by hash(runoob_id) buckets 8',
  'database-name'                     = 'test_cdc',
  'jdbc-url'                          = 'jdbc:mysql://172.16.**.**:9030',
  'load-url'                          = '172.16.**.**:18030',
  'table-name'                        = 'runoob_tbl',
  'username'                          = 'test',
  'password'                          = '123456',
  'sink.semantic'                     = 'exactly-once',
  'sink.properties.row_delimiter'     = '\x02',
  'sink.properties.column_separator'  = '\x01'
)
AS TABLE mysql.test_cdc.runoob_tbl
/*+ OPTIONS (
  'connector'     = 'mysql-cdc',
  'hostname'      = 'rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com',
  'port'          = '3306',
  'username'      = 'test',
  'password'      = '123456',
  'database-name' = 'test_cdc',
  'table-name'    = 'runoob_tbl'
) */;

Untuk opsi konfigurasi checkpoint, lihat Checkpointing.

Mode simple (direkomendasikan untuk penyiapan cepat)

Flink menurunkan definisi tabel dari skema MySQL — tidak perlu menentukan engine, kunci, atau distribusi secara manual. Tabel terpartisi tidak didukung dalam mode simple; buat partisi menggunakan mode normal sebagai gantinya.

/* Mode simple */

USE CATALOG sr;

CREATE TABLE IF NOT EXISTS runoob_tbl1 WITH (
  'starrocks.create.table.properties' = 'buckets 8',
  'starrocks.create.table.mode'       = 'simple',
  'database-name'                     = 'test_cdc',
  'jdbc-url'                          = 'jdbc:mysql://172.16.**.**:9030',
  'load-url'                          = '172.16.**.**:18030',
  'table-name'                        = 'runoob_tbl_sr',
  'username'                          = 'test',
  'password'                          = '123456',
  'sink.buffer-flush.interval-ms'     = '5000',
  'sink.properties.row_delimiter'     = '\x02',
  'sink.properties.column_separator'  = '\x01'
)
AS TABLE mysql.test_cdc.runoob_tbl
/*+ OPTIONS (
  'connector'     = 'mysql-cdc',
  'hostname'      = 'rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com',
  'port'          = '3306',
  'username'      = 'test',
  'password'      = '123456',
  'database-name' = 'test_cdc',
  'table-name'    = 'runoob_tbl'
) */;

Parameter klausa WITH

Jika versi Flink Anda lebih lama dari vvr-6.0.5-flink-1.15, tambahkan 'sink.use.new-apiapi' = 'false' ke klausa WITH. Untuk parameter sink tambahan, lihat Continuously load data from Apache Flink.

Parameter klausa OPTIONS

Klausa OPTIONS mengonfigurasi sumber CDC MySQL.

Publikasikan dan mulai penerapan

1. Pada tab Advanced di halaman Draft Editor, atur Engine Version ke vvr-6.0.3-flink-1.15 atau lebih baru.

2. Di pojok kanan atas halaman Draft Editor, klik Publish.

3. Pada halaman Deployments, temukan penerapan baru dan klik Start di kolom Actions.

Verifikasi data awal

Kueri tabel StarRocks untuk memastikan baris seed telah disinkronkan:

SELECT * FROM runoob_tbl1;

Output yang diharapkan:

+-----------+--------------+---------------+-----------------+---------+
| runoob_id | runoob_title | runoob_author | submission_date | add_col |
+-----------+--------------+---------------+-----------------+---------+
|        18 | first        | tom           | 2022-06-22      |       3 |
+-----------+--------------+---------------+-----------------+---------+

Verifikasi INSERT

Pada tab SQL Console instans ApsaraDB RDS for MySQL, masukkan satu baris:

INSERT INTO runoob_tbl (`runoob_id`, `runoob_title`, `runoob_author`, `submission_date`, `add_col`)
VALUES (1, 'second', 'tom2', '2022-06-23', 1);

Kueri StarRocks untuk memastikan kedua baris muncul:

SELECT * FROM runoob_tbl1;

Output yang diharapkan:

+-----------+--------------+---------------+-----------------+---------+
| runoob_id | runoob_title | runoob_author | submission_date | add_col |
+-----------+--------------+---------------+-----------------+---------+
|         1 | second       | tom2          | 2022-06-23      |       1 |
|        18 | new          | tom           | 2022-06-22      |       3 |
+-----------+--------------+---------------+-----------------+---------+

Verifikasi UPDATE

Pada SQL Console MySQL, perbarui satu baris:

UPDATE runoob_tbl SET runoob_title = 'new' WHERE runoob_id = 18;

Kueri StarRocks untuk memastikan perubahannya:

SELECT * FROM runoob_tbl1;

Output yang diharapkan:

+-----------+--------------+---------------+-----------------+---------+
| runoob_id | runoob_title | runoob_author | submission_date | add_col |
+-----------+--------------+---------------+-----------------+---------+
|         1 | second       | tom2          | 2022-06-23      |       1 |
|        18 | new          | tom           | 2022-06-22      |       3 |
+-----------+--------------+---------------+-----------------+---------+

Verifikasi DELETE

Pada SQL Console MySQL, hapus satu baris:

DELETE FROM runoob_tbl WHERE runoob_id = 1;

Kueri StarRocks untuk memastikan baris tersebut dihapus:

SELECT * FROM runoob_tbl1;

Output yang diharapkan:

+-----------+--------------+---------------+-----------------+---------+
| runoob_id | runoob_title | runoob_author | submission_date | add_col |
+-----------+--------------+---------------+-----------------+---------+
|        18 | new          | tom           | 2022-06-22      |       3 |
+-----------+--------------+---------------+-----------------+---------+

Verifikasi perubahan skema: tambahkan kolom nullable

Pada SQL Console MySQL, tambahkan kolom nullable dan masukkan satu baris dengan nilai pada kolom tersebut:

ALTER TABLE `runoob_tbl` ADD COLUMN `add_col2` INT;

INSERT INTO runoob_tbl (`runoob_id`, `runoob_title`, `runoob_author`, `submission_date`, `add_col`, `add_col2`)
VALUES (1, 'second', 'tom2', '2022-06-23', 1, 2);

Kueri StarRocks untuk memastikan kolom baru muncul dan baris yang ada menampilkan NULL:

SELECT * FROM runoob_tbl1;

Output yang diharapkan:

+-----------+--------------+---------------+-----------------+---------+----------+
| runoob_id | runoob_title | runoob_author | submission_date | add_col | add_col2 |
+-----------+--------------+---------------+-----------------+---------+----------+
|        18 | new          | tom           | 2022-06-22      |       3 |     NULL |
|         1 | second       | tom2          | 2022-06-23      |       1 |        2 |
+-----------+--------------+---------------+-----------------+---------+----------+

Kolom add_col2 secara otomatis ditambahkan ke StarRocks ketika baris data pertama yang memuatnya tiba.