Realtime Compute for Apache Flink：CREATE DATABASE AS (CDAS)

Sinkronisasi data

Evolusi skema

Selama sinkronisasi data database, perubahan skema (seperti menambahkan kolom) dapat disebarkan ke sink. Kebijakan ini konsisten dengan pernyataan CTAS. Untuk informasi lebih lanjut, lihat Evolusi Skema.

Batasan sintaksis

• Debugging draf SQL yang berisi pernyataan CDAS tidak didukung.

• MiniBatch tidak didukung.

  Penting

  • Pastikan konfigurasi MiniBatch telah dihapus sebelum membuat draf SQL yang berisi pernyataan CTAS atau CDAS. Lakukan langkah-langkah berikut:

    1. Buka O&M > Configurations.

    2. Pilih tab Deployment Defaults.

    3. Di bagian Other Configuration, verifikasi bahwa konfigurasi MiniBatch telah dihapus.

  • Jika kesalahan dilaporkan saat membuat atau memulai penyebaran dari draf SQL, lihat Bagaimana cara memperbaiki kesalahan "Saat ini tidak mendukung penggabungan StreamExecMiniBatchAssigner type ExecNode dalam sintaksis CTAS/CDAS"?.

Sistem hulu dan hilir yang didukung

Tabel berikut mencantumkan sistem hulu dan hilir yang didukung oleh pernyataan CDAS:

Sinkronkan database

Deskripsi: Sinkronkan semua tabel dari database MySQL tpcds ke Hologres.

Prasyarat: Katalog berikut dibuat di ruang kerja Anda:

• Katalog Hologres bernama holo.

• Katalog MySQL bernama mysql.

Contoh Kode:

USE CATALOG holo;

CREATE DATABASE IF NOT EXISTS holo_tpcds  -- Buat database bernama holo_tpcds di Hologres.
WITH ('sink.parallelism' = '4') -- Secara opsional konfigurasikan opsi untuk database tujuan. Secara default, paralelisme sink untuk Hologres diatur ke 4.
AS DATABASE mysql.tpcds INCLUDING ALL TABLES  -- Sinkronkan semua tabel.
/*+ OPTIONS('server-id'='8001-8004') */ ; -- Secara opsional konfigurasikan opsi untuk tabel sumber CDC MySQL.

Sinkronkan data lintas shard database

Deskripsi: Instance MySQL memiliki beberapa shard database bernama dari order_db01 hingga order_db99. Setiap shard database berisi beberapa tabel, seperti order dan order_detail. Pernyataan CDAS dapat digunakan untuk menyinkronkan semua tabel, bersama dengan data dan perubahan skema, di shard database ini ke Hologres.

Solusi:

Gunakan ekspresi reguler untuk nama database (`order_db[0-9]+`) untuk mencocokkan semua shard database (order_db01 hingga order_db99) untuk sinkronisasi. Nama database dan tabel ditambahkan ke setiap tabel sink sebagai dua bidang tambahan.

Kunci utama tabel Hologres mencakup nama database, nama tabel, dan kolom kunci utama tabel sumber untuk memastikan keunikan.

Tidak perlu membuat tabel target terlebih dahulu.

Contoh Kode dan Hasil:

Tabel dengan nama identik di seluruh shard database digabungkan sebelum disinkronkan ke dalam satu tabel Hologres.

USE CATALOG holo;

CREATE DATABASE IF NOT EXISTS holo_order  -- Buat database bernama holo_order di Hologres. Database ini berisi semua tabel di shard database order_db dari instance MySQL.
WITH('sink.parallelism'='4')        -- Tentukan parameter database tujuan. Secara default, paralelisme setiap sink Hologres adalah 4. Pengaturan ini opsional.
AS DATABASE mysql.`order_db[0-9]+` INCLUDING ALL TABLES -- Sinkronkan data dari semua tabel di shard database order_db dari instance MySQL.
/*+OPTIONS('server-id'='8001-8004')*/;  -- Tentukan parameter tambahan untuk tabel sumber CDC MySQL. Pengaturan ini opsional.

Sinkronkan tabel baru

Deskripsi: Setelah pekerjaan yang menyinkronkan data melalui pernyataan CDAS dimulai, tabel baru ditambahkan dan perlu disinkronkan.

Solusi: Aktifkan deteksi tabel baru untuk pekerjaan dan mulai ulang dari titik simpanan untuk menangkap dan menyinkronkan tabel yang baru ditambahkan.

Batasan: Deteksi tabel baru didukung di VVR 8.0.6 atau lebih baru. Untuk mengaktifkan fitur ini, pastikan mode startup tabel sumber diatur ke initial.

Prosedur:

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

2. Dalam dialog, perluas bagian More Strategies, pilih opsi Stop With Savepoint, lalu klik OK.

3. Dalam draf SQL pekerjaan, tambahkan pernyataan berikut untuk mengaktifkan deteksi tabel baru:

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

4. Klik Deploy.

5. Pulihkan pekerjaan dari titik simpanan.

   1. Di halaman Deployments, klik nama deployment Anda.

   2. Di halaman detail penyebaran, klik tab State. Lalu, klik subtab History.

   3. Di daftar Savepoints, temukan titik simpanan yang dibuat saat pekerjaan dibatalkan.

   4. Pilih More > Start job from this savepoint di kolom Actions. Untuk informasi lebih lanjut, lihat Mulai Penyebaran.

Jalankan beberapa pernyataan CDAS dan CTAS

Deskripsi: Sinkronkan data dari database tpcds, database tpch, dan database shard user_db01 hingga user_db99 ke Hologres dalam satu pekerjaan.

Solusi: Gunakan pernyataan STATEMENT SET untuk mengelompokkan beberapa pernyataan CDAS dan CTAS. Solusi ini menggunakan kembali simpul sumber untuk membaca data dari tabel yang diperlukan. Ini sangat bermanfaat dengan sumber data CDC MySQL, mengurangi jumlah ID server, koneksi database, dan beban baca keseluruhan pada database MySQL.

Contoh Kode:

USE CATALOG holo;

BEGIN STATEMENT SET;

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

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

END;

Sinkronkan data di database ke Kafka melalui beberapa pernyataan CDAS

Deskripsi: Sinkronkan data di tabel dari beberapa database MySQL (tpcds dan tpch, dll.) ke Kafka.

Solusi: Saat menggunakan beberapa pernyataan CDAS untuk menyinkronkan data di database ke Kafka, tabel dengan nama identik mungkin ada di database yang berbeda. Untuk mencegah konflik topik, konfigurasikan opsi cdas.topic.pattern untuk menentukan pola nama topik. Anda dapat menggunakan placeholder {table-name}. Misalnya, menentukan 'cdas.topic.pattern'='dbname-{table-name}' menghasilkan data yang direplikasi dari tabel table1 di database db1 ke topik Kafka dbname-table1.

Contoh Kode:

USE CATALOG kafkaCatalog;

BEGIN STATEMENT SET;

-- Sinkronkan data dari database tpcds.
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 data dari database tpch.
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;

Dengan memperkenalkan Kafka sebagai lapisan perantara antara MySQL dan Flink, Anda dapat mengurangi beban pada MySQL. Untuk informasi lebih lanjut, lihat Sinkronkan Data dari Semua Tabel di Database MySQL ke Kafka.

FAQ

Kesalahan runtime

• Apa yang harus saya lakukan jika penyebaran tidak dapat dimulai?

• Apa yang harus saya lakukan jika penyebaran dimulai ulang setelah penyebaran dijalankan?

• Apa yang harus saya lakukan jika pesan kesalahan "akka.pattern.AskTimeoutException" muncul?

Kinerja pekerjaan

• Bagaimana cara memecahkan masalah tekanan balik?

• Apa yang harus saya lakukan jika efisiensi pembacaan data rendah dan tekanan balik ada saat data penuh dibaca dari tabel?

• Bagaimana cara memecahkan masalah latensi tinggi?

• Apa yang harus saya lakukan jika konsumsi data hulu tidak stabil?

Sinkronisasi data

• Mengapa tipe data kunci utama di tabel MySQL berubah dari BIGINT UNSIGNED menjadi DECIMAL ketika saya menggunakan tabel MySQL untuk membuat katalog di konsol pengembangan Realtime Compute for Apache Flink? Mengapa tipe data kunci utama berubah menjadi TEXT setelah saya menjalankan pernyataan CREATE TABLE AS untuk menyinkronkan data ke Hologres?

• Bagaimana cara memperbaiki kesalahan "Saat ini tidak mendukung penggabungan StreamExecMiniBatchAssigner type ExecNode dalam sintaksis CTAS/CDAS"?

• Mengapa saya tidak mendapatkan keluaran di tabel sink?

• Bagaimana cara memecahkan masalah pembacaan sumber Flink?

• Bagaimana cara memecahkan masalah bahwa Flink yang sepenuhnya dikelola tidak dapat menulis data ke tabel sink?

• Bagaimana cara memecahkan masalah ketidakakuratan data?