All Products
Search

This Product

    E-MapReduce:Sinkronisasi data MySQL ke StarRocks menggunakan pernyataan CTAS di Realtime Compute for Apache Flink

    Document Center

    E-MapReduce:Sinkronisasi data MySQL ke StarRocks menggunakan pernyataan CTAS di Realtime Compute for Apache Flink

    Last Updated:Aug 22, 2026

    Anda dapat menggunakan pernyataan CREATE TABLE AS SELECT (CTAS) di Realtime Compute for Apache Flink untuk menyinkronkan data dari instans ApsaraDB RDS for MySQL ke kluster EMR Serverless StarRocks secara real time, termasuk perubahan skema.

    Latar Belakang

    Anda dapat menyinkronkan data MySQL ke EMR Serverless StarRocks melalui pernyataan CTAS atau CDAS. CTAS memungkinkan sinkronisasi struktur dan data untuk satu tabel, sedangkan CDAS memungkinkan sinkronisasi seluruh database atau beberapa tabel dalam satu database beserta struktur dan datanya. Artikel ini menggunakan pernyataan CTAS; penggunaan pernyataan CDAS mirip dengan CTAS. Untuk informasi lebih lanjut, lihat Sinkronisasi MySQL ke StarRocks menggunakan CTAS di Flink.

    Dengan pernyataan CTAS (CREATE TABLE AS), Anda dapat secara otomatis membuat tabel di StarRocks yang memiliki struktur identik dengan tabel di MySQL sekaligus melakukan sinkronisasi data. Selain itu, perubahan skema pada tabel sumber juga dapat disinkronkan secara real time ke tabel tujuan, sehingga meningkatkan efisiensi pembuatan tabel di penyimpanan target dan pemeliharaan perubahan struktur tabel sumber.

    Saat mengeksekusi pernyataan CTAS, Flink akan menjalankan alur kerja berikut:

    1. Memeriksa apakah tabel tujuan sudah ada di penyimpanan target.

      • Jika tidak ada, Flink akan membuat tabel tujuan di penyimpanan target melalui Catalog tujuan dengan skema yang sama seperti sumber data.

      • Jika sudah ada, proses pembuatan tabel dilewati. Jika tabel tujuan yang sudah ada memiliki skema yang tidak sesuai dengan tabel sumber, sistem akan mengembalikan error.

    2. Mengirimkan dan menjalankan pekerjaan sinkronisasi data yang sesuai. Data dari sumber serta perubahan skemanya akan disinkronkan ke tabel tujuan.

    Strategi sinkronisasi perubahan struktur tabel melalui pernyataan CTAS memungkinkan sinkronisasi perubahan skema tabel sumber ke tabel tujuan secara real time selama proses sinkronisasi data berlangsung.

    Perubahan skema mencakup pembuatan tabel awal serta perubahan tabel di masa depan.

    • Perubahan skema yang saat ini didukung untuk disinkronkan:

      • Menambahkan kolom nullable: Kolom yang sesuai akan ditambahkan secara otomatis di akhir skema tabel tujuan, dan data untuk kolom baru tersebut juga akan disinkronkan secara otomatis.

      • Menghapus kolom nullable: Kolom tersebut tidak langsung dihapus dari tabel tujuan, melainkan datanya diisi otomatis dengan nilai NULL.

      • Mengganti nama kolom: Kolom dengan nama baru akan ditambahkan di akhir tabel tujuan, dan data kolom lama akan diisi otomatis dengan nilai NULL.

        Contohnya, jika col_a diganti namanya menjadi col_b, maka col_b akan ditambahkan di akhir tabel tujuan, dan data col_a akan diisi otomatis dengan nilai NULL.

    • Perubahan skema yang belum didukung untuk disinkronkan:

      • Perubahan tipe data.

        Contohnya, perubahan dari VARCHAR ke BIGINT atau perubahan atribut dari NOT NULL menjadi NULLABLE.

      • Perubahan constraint seperti primary key atau indeks.

      • Penambahan atau penghapusan kolom non-null.

    Catatan
    • Jika skema tabel sumber mengalami salah satu perubahan di atas, Anda harus menghapus tabel tujuan dan me-restart pekerjaan yang mengeksekusi pernyataan CTAS. Dengan demikian, tabel tujuan akan dibuat ulang dan data historis akan disinkronkan kembali ke tabel tujuan.
    • Pernyataan CTAS tidak mengenali jenis pernyataan DDL, melainkan membandingkan perbedaan skema antara dua catatan data sebelum dan sesudah perubahan skema. Oleh karena itu, jika Anda menghapus sebuah kolom lalu menambahkannya kembali tanpa terjadi perubahan data di antara dua pernyataan DDL tersebut, CTAS akan menganggap tidak ada perubahan skema. Demikian pula, CTAS tidak akan memicu sinkronisasi perubahan skema meskipun Anda menambahkan kolom ke tabel sumber. Pernyataan ini hanya mengenali perubahan skema ketika terjadi perubahan data di tabel sumber, dan pada saat itulah perubahan skema disinkronkan ke tabel tujuan.
    • Untuk informasi lebih lanjut tentang tipe field yang didukung oleh pernyataan CTAS, lihat Continuously load data from Apache Flink®.

    Prasyarat

    Catatan

    Topik ini menggunakan MySQL 5.7 dan Flink vvr-8.0.11-flink-1.17 sebagai contoh.

    Batasan

    • Kluster Flink, instans EMR Serverless StarRocks, dan instans ApsaraDB RDS for MySQL harus berada dalam Virtual Private Cloud (VPC) yang sama.

    • Instans ApsaraDB RDS for MySQL harus menggunakan versi 5.7 atau lebih baru.

    Langkah 1: Siapkan data uji

    1. Buat database dan akun uji. Untuk informasi lebih lanjut, lihat Buat database dan akun.

      Setelah membuat database dan akun, berikan izin baca dan tulis kepada akun uji tersebut.

      Catatan

      Dalam contoh ini, database diberi nama test_cdc dan akun diberi nama test.

    2. Gunakan akun uji untuk terhubung ke instans MySQL. Untuk informasi lebih lanjut, lihat Masuk ke instans ApsaraDB RDS for MySQL menggunakan DMS.

    3. Jalankan perintah berikut di MySQL untuk membuat tabel data.

      use test_cdc;
-- Buat tabel
CREATE TABLE IF NOT EXISTS `runoob_tbl`(
   `runoob_id` INT UNSIGNED AUTO_INCREMENT,
   `runoob_title` VARCHAR(100) NOT NULL,
   `runoob_author` VARCHAR(40) NOT NULL,
   `submission_date` DATE,
   `add_col` int DEFAULT NULL,
   PRIMARY KEY ( `runoob_id` )
)ENGINE=InnoDB DEFAULT CHARSET=utf8;

-- Masukkan data
INSERT INTO test_cdc.`runoob_tbl` (`runoob_id`,`runoob_title`,`runoob_author`,`submission_date`,`add_col`) values (18,'first','tom','2025-06-22 17:13:44',3)

    4. Masuk dan terhubung ke instans EMR Serverless StarRocks. Untuk informasi lebih lanjut, lihat Terhubung ke instans StarRocks melalui client MySQL.

    5. Jalankan perintah berikut untuk membuat database test_cdc, membuat pengguna bernama test dengan password 1qaz!QAZ, serta memberikan izin kepada pengguna tersebut atas database dan tabel-tabelnya. Untuk informasi lebih lanjut, lihat Kelola pengguna.

      -- Buat database
CREATE DATABASE test_cdc;
-- Buat pengguna
CREATE USER 'test' IDENTIFIED by '1qaz!QAZ';
-- Berikan izin database kepada pengguna
GRANT ALL on test_cdc to test;
-- Berikan izin tabel kepada pengguna
GRANT ALL ON ALL TABLES IN DATABASE test_cdc to test;

    Langkah 2: Buat katalog

    Di halaman Data Management pada konsol Realtime Compute for Apache Flink, buat katalog MySQL dan StarRocks. Untuk informasi lebih lanjut, lihat Data Management.

    Catatan

    Konfigurasi parameter berikut hanya sebagai contoh. Sesuaikan nilainya berdasarkan lingkungan Anda.

    • Katalog MySQL

      • Kode contoh

        CREATE CATALOG mysql WITH (
  'type' = 'mysql',
  'hostname' = 'rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com',
  'port' = '3306',
  'username' = 'emr-test',
  'password' = '123456',
  'default-database' = 'test_cdc'
);

      • Parameter

        Parameter

        Deskripsi

        type

        Jenis katalog. Atur ke mysql.

        hostname

        Titik akhir internal instans ApsaraDB RDS for MySQL, yang dapat Anda salin dari halaman Koneksi Database di konsol ApsaraDB RDS for MySQL. Contoh: rm-2zepd6e20u3od****.mysql.rds.aliyuncs.com.

        port

        Nomor port layanan database MySQL. Nilai default-nya adalah 3306.

        username

        Username untuk layanan database MySQL.

        Gunakan username akun yang Anda buat di Langkah 1: Siapkan data uji. Dalam contoh ini, username-nya adalah test.

        password

        Password untuk layanan database MySQL.

        Gunakan password akun yang Anda buat di Langkah 1: Siapkan data uji.

        default-database

        Nama database MySQL default.

        Gunakan nama database yang Anda buat di Langkah 1: Siapkan data uji. Dalam contoh ini, namanya adalah test_cdc.

    • Katalog StarRocks

      • Kode contoh

        CREATE CATALOG sr  WITH (
  'type' = 'starrocks',
  'endpoint' = 'fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030',
  'username' = 'test',
  'password' = '1qaz!QAZ',
  'dbname' = 'test_cdc'
);

      • Parameter

        Parameter

        Deskripsi

        type

        Jenis katalog. Atur ke starrocks.

        endpoint

        Titik akhir internal dan port kueri node FE, dalam format Titik akhir internal node FE instans EMR Serverless StarRocks:9030.

        Contoh: fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030.

        Catatan

        Untuk mendapatkan titik akhir internal node FE, lihat Lihat daftar dan detail instans.

        username

        Username untuk kluster StarRocks.

        Gunakan username akun yang Anda buat di Langkah 1: Siapkan data uji. Dalam contoh ini, username-nya adalah test.

        password

        Password untuk layanan database StarRocks.

        Gunakan password akun yang Anda buat di Langkah 1: Siapkan data uji.

        dbname

        Nama database StarRocks.

        Gunakan nama database yang Anda buat di Langkah 1: Siapkan data uji. Dalam contoh ini, namanya adalah test_cdc.

    Langkah 3: Buat dan publikasikan pekerjaan

    1. Di halaman Data Development > ETL pada konsol Realtime Compute for Apache Flink, tulis pernyataan CTAS.

      Contoh berikut menunjukkan tiga mode berbeda:

      • Semantik at-least-once: Menggunakan parameter sink.buffer-flush.interval-ms untuk mengontrol interval penulisan ke StarRocks, menawarkan interval penulisan pendek dan penggunaan memori rendah.

        /*
      Semantik at-least-once
*/

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://fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030',
'load-url'='fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:8030',
'table-name'='runoob_tbl',
'username'='test',
'password' = '1qaz!QAZ',
'sink.buffer-flush.interval-ms' = '5000',
'sink.properties.row_delimiter' = '\x02',
'sink.properties.column_separator' = '\x01'
)
 as table mysql.test_cdc.runoob_tbl;

      • Semantik exactly-once: Memerlukan interval checkpoint. Menjamin tidak ada kehilangan atau duplikasi data, bahkan saat terjadi exception. Visibilitas data bergantung pada interval checkpoint. Untuk informasi lebih lanjut, lihat Checkpointing.

        /*
      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://fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030',
'load-url'='fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:8030',
'table-name'='runoob_tbl',
'username'='test',
'password' = '1qaz!QAZ',
'sink.semantic' = 'exactly-once',
'sink.properties.row_delimiter' = '\x02',
'sink.properties.column_separator' = '\x01'
)
 as table mysql.test_cdc.runoob_tbl;

      • Mode simple: Secara otomatis mencerminkan skema tabel MySQL sumber, sehingga tidak perlu menentukan field secara eksplisit. Tidak mendukung partisi. Gunakan mode normal untuk tabel yang memerlukan partisi.

        /*
      Dua contoh di atas menggunakan mode normal.
      Contoh ini menunjukkan mode simple.
*/

use CATALOG sr;

CREATE TABLE IF NOT EXISTS runoob_tbl with (
'starrocks.create.table.mode'='simple',
 'database-name'='test_cdc',
'jdbc-url'='jdbc:mysql://fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030',
'load-url'='fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:8030',
'table-name'='runoob_tbl',
'username'='test',
'password' = '1qaz!QAZ',
'sink.buffer-flush.interval-ms' = '5000',
'sink.properties.row_delimiter' = '\x02',
'sink.properties.column_separator' = '\x01'
)
 as table mysql.test_cdc.runoob_tbl;

      Tabel 1. Parameter WITH

      Parameter

      Wajib

      Deskripsi

      starrocks.create.table.properties

      Ya

      Klausa tambahan untuk pernyataan StarRocks CREATE TABLE, tidak termasuk definisi kolom. Contohnya termasuk engine, key, dan buckets.

      database-name

      Ya

      Nama database StarRocks.

      Dalam contoh ini, namanya adalah test_cdc.

      jdbc-url

      Ya

      URL JDBC yang digunakan untuk menjalankan operasi kueri di StarRocks.

      Contohnya, jdbc:mysql://fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030. Dalam string ini, fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com adalah titik akhir internal node FE instans EMR Serverless StarRocks.

      Catatan

      Untuk mendapatkan titik akhir internal node FE, lihat Lihat daftar dan detail instans.

      load-url

      Ya

      Tentukan titik akhir internal dan port kueri FE dalam format Titik akhir internal node FE instans EMR Serverless StarRocks:8030.

      Contoh: fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:8030.

      Catatan

      Untuk mendapatkan titik akhir internal node FE, lihat Lihat daftar dan detail instans.

      sink.semantic

      Tidak

      Jaminan pengiriman. Atur ke exactly-once untuk memastikan konsistensi data. Default: at-least-once.

      starrocks.create.table.mode

      Tidak

      Mode pembuatan tabel. Nilai yang valid:

      • mode normal (default): Anda harus menentukan properti lengkap, seperti engine, key, dan buckets, dalam konfigurasi starrocks.create.table.properties.

      • mode simple: Engine default-nya adalah olap dan tipe key-nya adalah primary key. Primary key harus identik dengan primary key tabel MySQL. Secara default, tabel didistribusikan berdasarkan hash pada semua kolom primary key dan tidak dipartisi. Dalam konfigurasi starrocks.create.table.properties, Anda harus menentukan parameter buckets. Parameter lain seperti properties bersifat opsional.

      sink.properties.row_delimiter

      Tidak

      Pemisah baris kustom.

      sink.properties.column_separator

      Tidak

      Pemisah kolom kustom.
      Catatan

      • Parameter sink.use.new-api telah dihapus pada versi Flink 1.15-vvr-6.0.5 dan yang lebih baru. Jika Anda menggunakan versi sebelum 1.15-vvr-6.0.5, Anda harus menambahkan 'sink.use.new-api'='false', ke parameter WITH.

      • Untuk informasi tentang konfigurasi lainnya, lihat Continuously load data from Apache Flink.

      Tabel 2. Opsi konektor sumber

      Parameter

      Deskripsi

      connector

      Jenis konektor. Atur ke mysql-cdc.

      hostname

      Titik akhir internal instans ApsaraDB RDS for MySQL. Anda dapat menyalinnya dari halaman Koneksi Database di konsol ApsaraDB RDS for MySQL. Contoh: rm-bp1nu0c46fn9k****.mysql.rds.aliyuncs.com.

      port

      Nomor port layanan database MySQL. Nilai default-nya adalah 3306.

      username

      Username untuk layanan database MySQL.

      Gunakan username akun yang Anda buat di Langkah 1: Siapkan data uji. Dalam contoh ini, username-nya adalah test.

      password

      Password untuk layanan database MySQL.

      Gunakan password akun yang Anda buat di Langkah 1: Siapkan data uji.

      table-name

      Nama tabel di StarRocks.

      Gunakan nama tabel yang Anda buat di Langkah 1: Siapkan data uji. Dalam contoh ini, namanya adalah runoob_tbl.

      database-name

      Nama database MySQL default.

      Gunakan nama database yang Anda buat di Langkah 1: Siapkan data uji. Dalam contoh ini, namanya adalah test_cdc.

    2. Klik Publish.

    3. Di halaman Publish new version, pilih target penerapan dan klik OK.

    4. Di halaman Job O&M, temukan pekerjaan target dan klik START di kolom Actions.

    Catatan

    Konsol Realtime Compute for Apache Flink tidak mendukung debugging pernyataan CTAS.

    Langkah 4: Verifikasi sinkronisasi data

    Kueri data

    1. Masuk dan terhubung ke instans EMR Serverless StarRocks. Untuk informasi lebih lanjut, lihat Terhubung ke instans StarRocks melalui client MySQL.

    2. Di jendela koneksi StarRocks, jalankan perintah berikut untuk melihat data tabel.

      use test_cdc;
select * from runoob_tbl;

      Output berikut menunjukkan bahwa data dari MySQL telah disinkronkan ke StarRocks.

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

    Penyisipan data

    1. Di jendela ApsaraDB RDS for MySQL, jalankan perintah berikut untuk menyisipkan data.

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

    2. Di jendela koneksi StarRocks, jalankan perintah berikut untuk melihat data tabel.

      select * from runoob_tbl;

      Output berikut menunjukkan bahwa data berhasil disisipkan.

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

    Pembaruan data

    1. Di jendela ApsaraDB RDS for MySQL, jalankan perintah berikut untuk memperbarui catatan.

      update runoob_tbl set runoob_title= 'new' where runoob_id = 18

    2. Di jendela koneksi StarRocks, jalankan perintah berikut untuk melihat data tabel.

      select * from runoob_tbl;

      Output berikut menunjukkan bahwa pembaruan data telah disinkronkan.

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

    Penghapusan data

    1. Di jendela ApsaraDB RDS for MySQL, jalankan perintah berikut untuk menghapus catatan.

      DELETE FROM runoob_tbl WHERE runoob_id = 1

    2. Di jendela koneksi StarRocks, jalankan perintah berikut untuk melihat data tabel.

      select * from runoob_tbl;

      Output berikut menunjukkan bahwa penghapusan data telah disinkronkan.

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

    Penambahan kolom nullable

    1. Di jendela ApsaraDB RDS for MySQL, jalankan perintah berikut untuk menambahkan kolom nullable.

      alter table `runoob_tbl` add COLUMN `add_col2` INT;

    2. Jalankan perintah berikut untuk menyisipkan data.

      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)

    3. Di jendela koneksi StarRocks, jalankan perintah berikut untuk melihat data tabel.

      select * from runoob_tbl;

      Output berikut menunjukkan bahwa perubahan skema berhasil disinkronkan.

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

    CDAS

    Pernyataan CDAS menyinkronkan seluruh database MySQL ke beberapa tabel yang sesuai di StarRocks. Anda dapat menggunakan klausa including table untuk memilih tabel tertentu yang akan disinkronkan.

    Seperti halnya CTAS, Anda harus membuat katalog MySQL dan StarRocks sebelum menjalankan pernyataan CDAS.

    CREATE DATABASE IF NOT EXISTS sr_db with (
'starrocks.create.table.properties'=' buckets 8',
'starrocks.create.table.mode'='simple',
'jdbc-url'='jdbc:mysql://fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:9030',
'load-url'='fe-c-9b354c83e891****-internal.starrocks.aliyuncs.com:8030',
'sink.buffer-flush.interval-ms' = '5000',
'sink.properties.row_delimiter' = '\x02',
'sink.properties.column_separator' = '\x01'
)
 as DATABASE mysql.test_cdc including table
 'tabl1','tbl2','tbl3';

    Referensi

    Anda juga dapat menyinkronkan data ke StarRocks menggunakan file YAML ingesti data. Untuk informasi lebih lanjut, lihat Ingesti Data.