Realtime Compute for Apache Flink:Pernyataan CREATE TABLE AS (CTAS) (Akan Dipensiunkan)

Sinkronisasi data

Sinkronisasi perubahan skema

Saat menyinkronkan data secara real-time, pernyataan CTAS juga mereplikasi perubahan skema dari tabel sumber ke tabel sink. Perubahan skema mencakup pembuatan tabel awal dan setiap perubahan tabel berikutnya.

• Perubahan skema yang didukung

  Perubahan skema	Deskripsi
  Tambah kolom nullable	Kolom yang sesuai secara otomatis ditambahkan di akhir skema tabel sink, dan datanya disinkronkan. Kolom baru secara default bersifat nullable, dan nilainya untuk baris yang sudah ada diatur ke NULL.
  Tambah kolom non-null	Kolom yang sesuai secara otomatis ditambahkan di akhir skema tabel sink, dan datanya disinkronkan.
  Hapus kolom nullable	Kolom tersebut tidak dihapus dari tabel sink. Sebagai gantinya, datanya secara otomatis diatur ke NULL.
  Ubah nama kolom	Hal ini dianggap sebagai penambahan kolom baru dan penghapusan kolom lama. Kolom yang diubah namanya ditambahkan di akhir skema tabel sink, dan data di kolom asli secara otomatis diatur ke NULL. Catatan Sebagai contoh, jika col_a diubah namanya menjadi col_b, maka col_b ditambahkan di akhir tabel sink, dan data untuk col_a diatur ke NULL.
  Ubah tipe data kolom	Jika sistem sink mendukung perubahan tipe data: Saat ini, hanya Paimon yang mendukung penanganan perubahan tipe data. CTAS mendukung perubahan tipe untuk kolom standar, seperti dari INT ke BIGINT. Perubahan tipe yang didukung bergantung pada aturan sink. Lihat dokumentasi konektor sink tertentu untuk detail selengkapnya. Jika sistem sink tidak mendukung perubahan tipe data: Saat ini, hanya Hologres yang mendukung mode normalisasi tipe untuk menangani perubahan tipe data. Dalam mode ini, pekerjaan CTAS membuat tabel downstream dengan tipe data yang lebih luas. Pendekatan ini memanfaatkan kompatibilitas sink terhadap perubahan tipe data. Untuk detail selengkapnya, lihat Contoh: Sinkronisasi data dalam mode normalisasi tipe. Penting Anda harus mengaktifkan mode normalisasi tipe saat pekerjaan CTAS pertama kali dijalankan. Jika tidak, Anda harus menghapus tabel sink dan melakukan restart tanpa status agar mode ini berlaku.

  Penting

  CTAS mendeteksi perubahan skema dengan membandingkan skema catatan berturut-turut, bukan dengan mengidentifikasi operasi DDL tertentu.

  • Jika sebuah kolom dihapus lalu ditambahkan kembali tanpa perubahan data di antaranya, CTAS tidak mendeteksi adanya perubahan skema.

  • CTAS hanya mendeteksi perubahan skema setelah kolom baru ditambahkan dan terjadi perubahan data. Kemudian, perubahan skema tersebut disinkronkan ke tabel sink.

• Perubahan skema yang tidak didukung

  • Perubahan pada constraint, seperti primary key atau indeks.

  • Penghapusan kolom non-null.

  • Mengubah kolom dari NOT NULL menjadi NULLABLE.

  Penting

  Jika terjadi perubahan skema yang tidak didukung, Anda harus secara manual menghapus tabel sink dan me-restart pekerjaan CTAS. Hal ini akan membuat ulang tabel sink dan menyinkronkan ulang data historis.

Sinkronisasi tabel tunggal

Skenario: Menyinkronkan tabel web_sales dari MySQL ke Hologres.

Prasyarat: Anda telah mendaftarkan katalog berikut di ruang kerja Anda.

• Katalog Hologres bernama holo.

• Katalog MySQL bernama mysql.

Contoh kode:

CTAS biasanya digunakan dengan katalog baik untuk sumber data maupun tujuan. Katalog sumber secara otomatis mengurai skema dan opsi tabel sumber, sehingga menghilangkan kebutuhan DDL manual dan memungkinkan sinkronisasi data historis serta inkremental dari tabel sumber ke tabel sink.

USE CATALOG holo;

CREATE TABLE IF NOT EXISTS web_sales   -- Jika database tidak ditentukan, tabel disinkronkan ke tabel web_sales di database default.
WITH ('jdbcWriteBatchSize' = '1024')   -- Opsional: Menentukan parameter untuk tabel sink.
AS TABLE mysql.tpcds.web_sales   
/*+ OPTIONS('server-id'='8001-8004') */;  -- Menentukan parameter tambahan untuk tabel sumber mysql-cdc.

Gabung tabel terpartisi

Skenario: Gunakan CTAS untuk menggabungkan beberapa tabel MySQL terpartisi menjadi satu tabel Hologres.

Solusi: Dalam kombinasi dengan katalog MySQL, gunakan regular expression untuk mencocokkan nama database dan tabel yang ingin Anda sinkronkan. Nama database dan tabel ditulis sebagai dua kolom tambahan di tabel sink. Untuk memastikan keunikan primary key, nama database, nama tabel, dan primary key asli digabungkan untuk membentuk primary key komposit baru untuk tabel Hologres.

Contoh kode dan hasil penggabungan:

USE CATALOG holo;

CREATE TABLE IF NOT EXISTS user
WITH ('jdbcWriteBatchSize' = '1024')
AS TABLE mysql.`wp.*`.`user[0-9]+`  
/*+ OPTIONS('server-id'='8001-8004') */;

ALTER TABLE `user02` ADD COLUMN `age` INT;
INSERT INTO `user02` (id, name, age) VALUES (27, 'Tony', 30);

Sinkronisasi kolom terhitung

Skenario: Tambahkan kolom terhitung kustom ke tabel Hologres selama proses penggabungan dan sinkronisasi tabel MySQL terpartisi.

Contoh kode dan hasil penggabungan:

USE CATALOG holo;

CREATE TABLE IF NOT EXISTS user
WITH ('jdbcWriteBatchSize' = '1024')
AS TABLE mysql.`wp.*`.`user[0-9]+`
/*+ OPTIONS('server-id'='8001-8004') */
ADD COLUMN (
  `c_id` AS `id` + 10 AFTER `id`,
  `calss` AS 3  AFTER `id`
);

Pernyataan CTAS ganda sebagai satu pekerjaan

Skenario: Menyinkronkan tabel web_sales MySQL dan tabel user terpartisi ke Hologres sebagai satu pekerjaan.

Solusi: Gunakan sintaks STATEMENT SET untuk mengirim beberapa pernyataan CTAS sebagai satu pekerjaan. Pendekatan ini memungkinkan satu node sumber digunakan ulang untuk membaca data dari beberapa tabel bisnis, sehingga mengurangi jumlah server ID, koneksi database, dan beban baca pada sumber MySQL CDC.

Contoh kode:

USE CATALOG holo;

BEGIN STATEMENT SET;

-- Sinkronisasi tabel web_sales.
CREATE TABLE IF NOT EXISTS web_sales
AS TABLE mysql.tpcds.web_sales
/*+ OPTIONS('server-id'='8001-8004') */;

-- Sinkronisasi tabel user terpartisi.
CREATE TABLE IF NOT EXISTS user
AS TABLE mysql.`wp.*`.`user[0-9]+`
/*+ OPTIONS('server-id'='8001-8004') */;

END;

Sinkronisasi satu sumber ke beberapa tabel sink

• Jika tabel sink tidak memerlukan kolom terhitung

  USE CATALOG `holo`;
  
  BEGIN STATEMENT SET;
  
  -- Gunakan pernyataan CTAS untuk menyinkronkan tabel user MySQL ke tabel user di database1 gudang data Hologres.
  CREATE TABLE IF NOT EXISTS `database1`.`user`
  AS TABLE `mysql`.`tpcds`.`user`
  /*+ OPTIONS('server-id'='8001-8004') */;
  
  -- Gunakan pernyataan CTAS untuk menyinkronkan tabel user MySQL ke tabel user di database2 gudang data Hologres.
  CREATE TABLE IF NOT EXISTS `database2`.`user`
  AS TABLE `mysql`.`tpcds`.`user`
  /*+ OPTIONS('server-id'='8001-8004') */;
  
  END;

• Jika tabel sink memerlukan kolom terhitung

  -- Buat tabel temporary user_with_changed_id berdasarkan tabel sumber user. Mendukung definisi kolom terhitung, seperti computed_id, yang dihitung berdasarkan id dari tabel sumber.
  CREATE TEMPORARY TABLE `user_with_changed_id` (
    `computed_id` AS `id` + 1000
  ) LIKE `mysql`.`tpcds`.`user`;
  
  -- Buat tabel temporary user_with_changed_age berdasarkan tabel sumber user. Mendukung definisi kolom terhitung, seperti computed_age, yang dihitung berdasarkan age dari tabel sumber.
  CREATE TEMPORARY TABLE `user_with_changed_age` (
    `computed_age` AS `age` + 1
  ) LIKE `mysql`.`tpcds`.`user`;
  
  BEGIN STATEMENT SET;
  
  -- Gunakan pernyataan CTAS untuk menyinkronkan tabel user MySQL ke tabel user_with_changed_id di gudang data Hologres. Tabel ini akan berisi ID terhitung di kolom computed_id. 
  CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`user_with_changed_id`
  AS TABLE `user_with_changed_id`
  /*+ OPTIONS('server-id'='8001-8004') */;
  
  -- Gunakan pernyataan CTAS untuk menyinkronkan tabel user MySQL ke tabel user_with_changed_age di gudang data Hologres. Tabel ini akan berisi usia terhitung di kolom computed_age. 
  CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`user_with_changed_age`
  AS TABLE `user_with_changed_age`
  /*+ OPTIONS('server-id'='8001-8004') */;
  
  END;

Sinkronisasi tabel baru

Skenario: Setelah pekerjaan dengan beberapa pernyataan CTAS dijalankan, Anda perlu menambahkan pernyataan CTAS baru untuk menyinkronkan data dari tabel yang baru ditambahkan.

Solusi: Aktifkan fitur deteksi tabel baru di pekerjaan SQL, tambahkan pernyataan CTAS baru, lalu restart pekerjaan dari titik simpan. Setelah tabel baru terdeteksi, datanya akan disinkronkan.

Batasan:

• Fitur deteksi tabel baru didukung di VVR 8.0.1 dan versi lebih baru.

• Saat menyinkronkan dari tabel sumber CDC, fitur deteksi tabel baru hanya didukung untuk pekerjaan dengan mode startup tabel sumber diatur ke initial.

• Konfigurasi tabel sumber baru dalam pernyataan CTAS yang ditambahkan harus identik dengan konfigurasi tabel sumber yang sudah ada untuk memastikan sumber dapat digunakan ulang.

• Parameter konfigurasi pekerjaan, seperti mode startup, tidak boleh diubah sebelum dan sesudah penambahan pernyataan CTAS baru.

Prosedur:

1. Saat Anda perlu menambahkan pernyataan CTAS baru, buka halaman O&M (Operations and Maintenance), hentikan pekerjaan, dan pilih Stop With Savepoint.

2. Di pekerjaan SQL, aktifkan deteksi tabel baru, tambahkan pernyataan CTAS baru, lalu Deploy pekerjaan lagi.

   1. Tambahkan pernyataan berikut ke pekerjaan SQL untuk mengaktifkan deteksi tabel baru.

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

   2. Tambahkan pernyataan CTAS baru ke pekerjaan SQL. Kode lengkap akhirnya adalah sebagai berikut.

      -- Aktifkan deteksi tabel baru.
      SET 'table.cdas.scan.newly-added-table.enabled' = 'true';
      
      USE CATALOG holo;
      
      BEGIN STATEMENT SET;
      
      -- Sinkronisasi tabel web_sales.
      CREATE TABLE IF NOT EXISTS web_sales
      AS TABLE mysql.tpcds.web_sales
      /*+ OPTIONS('server-id'='8001-8004') */;
      
      -- Sinkronisasi tabel user terpartisi.
      CREATE TABLE IF NOT EXISTS user
      AS TABLE mysql.`wp.*`.`user[0-9]+`
      /*+ OPTIONS('server-id'='8001-8004') */;
      
      -- Sinkronisasi tabel product. (Tabel baru)
      CREATE TABLE IF NOT EXISTS product
      AS TABLE mysql.tpcds.product
      /*+ OPTIONS('server-id'='8001-8004') */;
      
      END;

   3. Klik Deploy.

3. Pulihkan pekerjaan dari titik simpan.

   1. Di halaman O&M (Operations and Maintenance), klik nama pekerjaan target, buka tab State Management, lalu klik History.

   2. Di daftar Savepoints, temukan titik simpan yang dibuat saat pekerjaan dihentikan.

   3. Di kolom Actions untuk titik simpan target, pilih More > Start Job from This Savepoint untuk menjalankan pekerjaan. Untuk informasi selengkapnya, lihat Jalankan pekerjaan.

Sinkronisasi ke tabel partisi Hologres

Skenario: Gunakan pernyataan CTAS untuk menyinkronkan tabel sumber MySQL ke tabel partisi Hologres.

Aturan partisi Hologres: Di Hologres, jika primary key ditentukan untuk tabel sink, kolom partisi harus termasuk dalam primary key.

Contoh kode:

Pernyataan DDL untuk tabel sumber MySQL adalah sebagai berikut:

CREATE TABLE orders (
    order_id INTEGER NOT NULL,
    product_id INTEGER NOT NULL,
    city VARCHAR(100) NOT NULL
    order_date DATE,
    purchaser INTEGER,
    PRIMARY KEY(order_id, product_id)
);

Pendekatannya berbeda tergantung pada apakah primary key tabel sumber mencakup kunci partisi.

• Jika primary key sumber mencakup kolom partisi, Anda dapat menyinkronkan langsung dengan pernyataan CTAS.

  Hologres secara otomatis memverifikasi bahwa kolom partisi merupakan bagian dari primary key.

  CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`orders`
  PARTITIONED BY (product_id)
  AS TABLE `mysql`.`tpcds`.`orders`;

• Jika primary key sumber tidak mencakup kolom partisi, deklarasikan ulang primary key tabel sink dalam pernyataan CTAS.

  Jika kolom partisi (misalnya, city) bukan bagian dari primary key tabel sumber, pekerjaan akan gagal. Anda harus mendeklarasikan ulang primary key tabel sink dalam pernyataan CTAS untuk memastikan kolom partisi termasuk dalam primary key.

  -- Anda dapat menggunakan SQL berikut untuk menentukan primary key tabel partisi Hologres sebagai order_id, product_id, dan city.
  CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`orders`(
      CONSTRAINT `PK_order_id_city` PRIMARY KEY (`order_id`,`product_id`,`city`) NOT ENFORCED
  )
  PARTITIONED BY (city)
  AS TABLE `mysql`.`tpcds`.`orders`;

Sinkronisasi data dalam mode normalisasi tipe

Skenario: Saat menggunakan pernyataan CTAS untuk menyinkronkan data ke tabel Hologres, Anda perlu mendukung skenario yang memerlukan penyesuaian presisi tipe data yang ada (misalnya, dari VARCHAR(10) ke VARCHAR(20)) atau mengubah tipe data (misalnya, dari SMALLINT ke INT).

Solusi: Gunakan mode normalisasi tipe Hologres untuk menyinkronkan data. Mode normalisasi tipe harus diaktifkan saat pekerjaan CTAS pertama kali dijalankan. Jika tidak diaktifkan saat startup, Anda harus menghapus tabel downstream dan melakukan restart tanpa status agar mode ini berlaku.

Aturan normalisasi tipe:

Setelah tipe data upstream diubah, jika tipe baru dan asli dinormalisasi ke tipe target yang sama, pekerjaan berjalan normal. Jika tidak, perubahan tersebut dianggap tidak kompatibel, dan pekerjaan CTAS melemparkan exception. Aturan spesifiknya adalah sebagai berikut:

• TINYINT, SMALLINT, INT, dan BIGINT dinormalisasi ke BIGINT.

• CHAR, VARCHAR, dan STRING dinormalisasi ke STRING.

• FLOAT dan DOUBLE dinormalisasi ke DOUBLE.

• Tipe data lainnya dibuat sesuai aturan pemetaan tipe asli. Untuk informasi selengkapnya, lihat Pemetaan tipe.

Contoh kode:

CREATE TABLE IF NOT EXISTS `holo`.`tpcds`.`orders` 
WITH (
'connector' = 'hologres', 
'enableTypeNormalization' = 'true' -- Aktifkan mode normalisasi tipe.
) AS TABLE `mysql`.`tpcds`.`orders`;

Sinkronisasi MongoDB ke Hologres

Batasan:

• Memerlukan Realtime Compute for Apache Flink VVR 8.0.6 atau versi lebih baru dan MongoDB 6.0 atau versi lebih baru.

• Di SQL Hints, atur scan.incremental.snapshot.enabled dan scan.full-changelog ke true.

• Fitur pre- dan post-images harus diaktifkan di database MongoDB. Untuk informasi cara mengaktifkannya, lihat Document Preimages.

• Menyinkronkan beberapa koleksi MongoDB dalam satu pekerjaan memiliki persyaratan berikut:

  • Konfigurasi MongoDB harus identik untuk setiap tabel, termasuk hosts, scheme, username, password, dan connectionOptions.

  • Konfigurasi scan.startup.mode harus identik untuk setiap tabel.

Contoh kode:

BEGIN STATEMENT SET;

CREATE TABLE IF NOT EXISTS `holo`.`database`.`table1`
AS TABLE `mongodb`.`database`.`collection1`
/*+ OPTIONS('scan.incremental.snapshot.enabled'='true','scan.full-changelog'='true') */;

CREATE TABLE IF NOT EXISTS `holo`.`database`.`table2`
AS TABLE `mongodb`.`database`.`collection2`
/*+ OPTIONS('scan.incremental.snapshot.enabled'='true','scan.full-changelog'='true') */;

END;