Realtime Compute for Apache Flink:Pernyataan CREATE DATABASE AS (CDAS)

Sinkronisasi data

Evolusi skema

Saat CDAS menyinkronkan database, perubahan skema—seperti penambahan kolom—secara otomatis disebarkan ke sink. Perilaku dan kebijakannya sama seperti pada CTAS. Lihat Evolusi skema.

Batasan sintaks

• Debugging draf SQL yang berisi pernyataan CDAS tidak didukung.

• MiniBatch tidak didukung.

  Penting

  Hapus semua konfigurasi MiniBatch sebelum membuat draf SQL dengan pernyataan CTAS atau CDAS: 1. Buka O&M > Configurations. 2. Pilih tab Deployment Defaults. 3. Di bagian Other Configuration, pastikan konfigurasi MiniBatch telah dihapus. Jika Anda menemui error "Currently does not support merge StreamExecMiniBatchAssigner type ExecNode in CTAS/CDAS syntax" saat membuat atau memulai deployment, lihat Bagaimana cara memperbaiki error ini?

Konektor yang didukung

Parameter

Sinkronkan database

Menyinkronkan semua tabel dari database MySQL tpcds ke Hologres.

Prasyarat:

• Katalog Hologres bernama holo telah dibuat di ruang kerja.

• Katalog MySQL bernama mysql telah dibuat di ruang kerja.

USE CATALOG holo;

CREATE DATABASE IF NOT EXISTS holo_tpcds          -- Buat holo_tpcds di Hologres.
WITH ('sink.parallelism' = '4')                   -- Atur parallelisme sink (default: 4 untuk Hologres).
AS DATABASE mysql.tpcds INCLUDING ALL TABLES      -- Sinkronkan semua tabel dari mysql.tpcds.
/*+ OPTIONS('server-id'='8001-8004') */;          -- Konfigurasi opsi sumber MySQL CDC.

Opsi yang ditetapkan dalam klausa WITH hanya berlaku untuk pekerjaan saat ini dan mengontrol perilaku penulisan. Opsi tersebut tidak dipertahankan di katalog Hologres. Untuk opsi yang didukung, lihat Hologres.

Konsolidasi dan sinkronkan shard database

Gabungkan semua tabel dengan nama identik di beberapa shard database MySQL menjadi tabel Hologres tunggal.

Skenario: Instans MySQL memiliki shard bernama order_db01 hingga order_db99. Setiap shard berisi tabel seperti order dan order_detail. Semua data shard—termasuk perubahan skema—harus masuk ke Hologres.

Solusi: Gunakan ekspresi reguler pada nama database untuk mencocokkan semua shard. Nama database dan tabel ditambahkan ke setiap tabel sink sebagai dua field tambahan. Kunci primer Hologres mencakup nama database, nama tabel, dan kolom kunci primer tabel sumber untuk memastikan keunikan. Tidak perlu membuat tabel target terlebih dahulu.

USE CATALOG holo;

CREATE DATABASE IF NOT EXISTS holo_order              -- Buat holo_order di Hologres.
WITH('sink.parallelism'='4')                          -- Atur parallelisme sink (opsional).
AS DATABASE mysql.`order_db[0-9]+` INCLUDING ALL TABLES  -- Cocokkan semua shard order_db.
/*+ OPTIONS('server-id'='8001-8004') */;              -- Konfigurasi opsi sumber MySQL CDC (opsional).

Tabel dengan nama identik di seluruh shard digabung menjadi satu tabel Hologres.

Sinkronkan tabel baru

Setelah pekerjaan CDAS yang sedang berjalan dimulai, aktifkan deteksi tabel baru dan restart dari titik simpan untuk menangkap tabel yang baru ditambahkan.

Deteksi tabel baru memerlukan VVR 8.0.6 atau lebih baru. Mode startup tabel sumber harus initial.

1. Di halaman Deployments, temukan deployment target dan klik Cancel di kolom Actions.

2. Di dialog, perluas More Strategies, pilih Stop With Savepoint, lalu klik OK.

3. Di draf SQL, tambahkan pernyataan berikut:

   SET 'table.cdas.scan.newly-added-table.enabled' = 'true';

4. Klik Deploy.

5. Pulihkan pekerjaan dari titik simpan:

   1. Di halaman Deployments, klik nama deployment.

   2. Di halaman detail deployment, klik tab State, lalu subtab History.

   3. Di daftar Savepoints, temukan titik simpan yang dibuat saat Anda menghentikan pekerjaan.

   4. Pilih More > Start job from this savepoint pada kolom Actions. Lihat Memulai pekerjaan.

Jalankan beberapa pernyataan CDAS dan CTAS sebagai satu pekerjaan

Menyinkronkan beberapa database dalam satu pekerjaan untuk memaksimalkan penggunaan ulang sumber.

Skenario: Sinkronkan tpcds, tpch, dan shard user_db01–user_db99 ke Hologres dalam satu pekerjaan.

Solusi: Bungkus semua pernyataan dalam blok STATEMENT SET. Flink menggunakan ulang satu vertex sumber di semua pernyataan, mengurangi jumlah server ID, koneksi database, dan beban baca keseluruhan—terutama untuk sumber MySQL CDC.

USE CATALOG holo;

BEGIN STATEMENT SET;

-- Sinkronkan tabel user dari semua shard user_db.
CREATE TABLE IF NOT EXISTS user
AS TABLE mysql.`user_db[0-9]+`.`user[0-9]+`
/*+ OPTIONS('server-id'='8001-8004') */;

-- Sinkronkan database tpcds.
CREATE DATABASE IF NOT EXISTS holo_tpcds
AS DATABASE mysql.tpcds INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */;

-- Sinkronkan database tpch.
CREATE DATABASE IF NOT EXISTS holo_tpch
AS DATABASE mysql.tpch INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */;

END;

Sinkronkan beberapa database MySQL ke Kafka

Menyinkronkan tabel dari beberapa database MySQL ke Kafka sambil menghindari konflik nama topik.

Skenario: Tabel dengan nama yang sama ada di tpcds dan tpch. Mengirim keduanya ke topik Kafka yang sama akan menimpa data.

Solusi: Gunakan opsi cdas.topic.pattern untuk menghasilkan nama topik unik per database. Placeholder {table-name} diganti saat runtime dengan nama tabel aktual. Misalnya, 'cdas.topic.pattern'='tpcds-{table-name}' mengarahkan table1 dari tpcds ke topik tpcds-table1.

USE CATALOG kafkaCatalog;

BEGIN STATEMENT SET;

-- Sinkronkan database tpcds. Nama topik mengikuti pola "tpcds-<table_name>".
CREATE DATABASE IF NOT EXISTS kafka
WITH ('cdas.topic.pattern' = 'tpcds-{table-name}')
AS DATABASE mysql.tpcds INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */;

-- Sinkronkan database tpch. Nama topik mengikuti pola "tpch-<table_name>".
CREATE DATABASE IF NOT EXISTS kafka
WITH ('cdas.topic.pattern' = 'tpch-{table-name}')
AS DATABASE mysql.tpch INCLUDING ALL TABLES
/*+ OPTIONS('server-id'='8001-8004') */;

END;

Menggunakan Kafka sebagai lapisan perantara antara MySQL dan Flink juga mengurangi beban pada MySQL. Lihat Sinkronkan database MySQL ke Kafka dengan Flink CDC.