Realtime Compute for Apache Flink：CREATE TABLE AS (CTAS)

Sinkronisasi data

Evolusi skema

Selama sinkronisasi data, pernyataan CTAS mendukung replikasi perubahan skema, termasuk pembuatan tabel dan modifikasi skema, dari tabel sumber ke tabel sink.

• Perubahan skema yang didukung

  Perubahan skema	Deskripsi
  Menambahkan kolom nullable	Secara otomatis menambahkan kolom terkait ke akhir skema tabel sink, dengan data yang disinkronkan ke kolom yang ditambahkan. Kolom baru secara default diatur sebagai kolom nullable, dan data pada kolom ini sebelum perubahan secara otomatis diatur ke NULL.
  Menambahkan kolom non-null	Secara otomatis menambahkan kolom yang sesuai ke akhir skema tabel sink dan menyinkronkan data.
  Menghapus kolom nullable	Secara otomatis mengisi nilai null pada kolom nullable di tabel sink alih-alih menghapus kolom tersebut dari tabel.
  Mengganti nama kolom	Operasi penggantian nama kolom melibatkan penambahan kolom dan penghapusan kolom. Setelah nama kolom diubah di tabel sumber, kolom dengan nama baru ditambahkan ke akhir tabel sink dan kolom dengan nama asli diisi dengan nilai null. Catatan Sebagai contoh, jika nama kolom col_a di tabel sumber diubah menjadi col_b, maka kolom col_b ditambahkan ke akhir tabel sink dan kolom col_a secara otomatis diisi dengan nilai null.
  Memodifikasi tipe data kolom	Untuk sistem hilir yang mendukung modifikasi tipe kolom: Saat ini, hanya Paimon yang mendukung perubahan tipe. Pernyataan CTAS mendukung modifikasi tipe kolom biasa, misalnya dari INT ke BIGINT. Kompatibilitas bergantung pada aturan spesifik sistem (lihat dokumentasi konektor). Untuk sistem hilir yang tidak mendukung modifikasi tipe kolom: Saat ini, hanya Hologres yang mendukung penggunaan pelebaran tipe (type widening) untuk menangani perubahan tipe kolom. Mekanismenya sebagai berikut: Saat startup pekerjaan, tabel Hologres dengan tipe data yang lebih luas dibuat, dan perubahan tipe kolom didukung berdasarkan kompatibilitas sink. Untuk informasi selengkapnya, lihat Contoh: Menyinkronkan data dalam mode normalisasi tipe. Penting Untuk mengaktifkan dukungan pelebaran tipe pada Hologres, aktifkan normalisasi tipe saat peluncuran pekerjaan awal. Untuk pekerjaan yang sudah ada, hapus tabel Hologres dan restart tanpa state untuk menerapkan pengaturan normalisasi tipe.

  Penting

  Saat pernyataan CTAS digunakan untuk sinkronisasi, sistem hanya membandingkan perbedaan skema dan tidak mengidentifikasi jenis DDL spesifik. Sebagai contoh:

  • Jika sebuah kolom dihapus lalu ditambahkan kembali tanpa perubahan data selama periode tersebut, sistem tidak mendeteksi adanya perubahan skema.

  • Jika sebuah kolom dihapus lalu ditambahkan kembali dengan perubahan data selama periode tersebut, sistem mendeteksi dan menyinkronkan perubahan skema.

• Perubahan skema yang tidak didukung

  • Memodifikasi constraint, seperti primary key atau indeks.

  • Menghapus kolom non-nullable.

  • Mengubah dari NOT NULL menjadi NULLABLE.

  Penting

  Untuk menyinkronkan perubahan skema yang tidak didukung, hapus secara manual tabel sink dan restart pekerjaan Anda untuk menyinkronkan ulang data historis.

Menyinkronkan tabel

Deskripsi: Menyinkronkan tabel web_sales dari MySQL ke Hologres.

Prasyarat:

• Katalog Hologres bernama holo telah dibuat.

• Katalog MySQL bernama mysql telah dibuat.

Kode contoh:

Pernyataan CTAS sering digunakan bersama katalog sumber dan tujuan untuk mendukung sinkronisasi data penuh dan inkremental. Katalog sumber secara otomatis mengurai skema dan properti tabel sumber tanpa DDL eksplisit.

USE CATALOG holo;

CREATE TABLE IF NOT EXISTS web_sales   -- Menyinkronkan data ke tabel web_sales di database default.
WITH ('jdbcWriteBatchSize' = '1024')   -- Secara opsional konfigurasikan opsi konektor untuk tabel sink.
AS TABLE mysql.tpcds.web_sales   
/*+ OPTIONS('server-id'='8001-8004') */; -- Secara opsional konfigurasikan opsi tambahan untuk tabel sumber CDC MySQL.

Menggabungkan dan menyinkronkan shard tabel dan database

Deskripsi: Menggabungkan tabel dan database MySQL yang di-shard sebelum menyinkronkan data ke tabel Hologres.

Metode: Gunakan katalog MySQL dan regular expressions untuk mencocokkan database dan tabel yang ingin disinkronkan.

Nama database dan tabel ditulis ke tabel sink sebagai nilai dua bidang tambahan. Primary key tabel sink terdiri dari nama database, nama tabel, dan primary key asli untuk memastikan keunikan primary key.

Kode dan hasil:

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);

Menyinkronkan kolom terhitung kustom

Deskripsi: Selama sinkronisasi gabungan shard tabel dan database dari MySQL ke Hologres, tambahkan kolom terhitung kustom.

Kode dan hasil:

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`
);

Menjalankan beberapa pernyataan CTAS dalam satu pekerjaan

Deskripsi: Menyinkronkan shard tabel web_sales dan user dari MySQL ke Hologres dalam satu pekerjaan.

Metode: Gunakan STATEMENT SET untuk mengeksekusi beberapa pernyataan CTAS sebagai satu kelompok. Pendekatan ini menggunakan kembali vertex sumber untuk membaca data dari beberapa tabel, sehingga mengurangi jumlah server ID, koneksi database, dan beban baca secara keseluruhan.

Kode contoh:

USE CATALOG holo;

BEGIN STATEMENT SET;

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

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

END;

Menyinkronkan data dari sumber ke beberapa sink menggunakan beberapa pernyataan CTAS

• Tidak ada kolom terhitung yang ditambahkan ke tabel sink

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

• Kolom terhitung ditambahkan ke tabel sink

  -- Buat tabel temporary bernama user_with_changed_id berdasarkan tabel sumber user. Definisikan kolom computed_id berdasarkan kolom id tabel sumber.
  CREATE TEMPORARY TABLE `user_with_changed_id` (
    `computed_id` AS `id` + 1000
  ) LIKE `mysql`.`tpcds`.`user`;
  
  -- Buat tabel temporary bernama user_with_changed_age berdasarkan tabel sumber user. Definisikan kolom computed_age berdasarkan kolom age tabel sumber.
  CREATE TEMPORARY TABLE `user_with_changed_age` (
    `computed_age` AS `age` + 1
  ) LIKE `mysql`.`tpcds`.`user`;
  
  BEGIN STATEMENT SET;
  
  -- Menyinkronkan data dari tabel database MySQL user ke tabel user_with_changed_id Hologres. Tabel user_with_changed_id berisi ID yang diperoleh dari perhitungan berdasarkan kolom id tabel sumber. ID tersebut berada 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') */;
  
  -- Menyinkronkan data dari tabel database MySQL user ke tabel user_with_changed_age Hologres. Tabel user_with_changed_age berisi nilai usia yang diperoleh dari perhitungan berdasarkan kolom age tabel sumber. Nilai usia tersebut berada 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;

Menyinkronkan tabel baru menggunakan beberapa pernyataan CTAS

Deskripsi skenario: Setelah pekerjaan yang menggunakan beberapa pernyataan CTAS untuk sinkronisasi dimulai, tambahkan pernyataan CTAS untuk mereplikasi tabel baru.

Metode: Aktifkan deteksi tabel baru untuk pekerjaan, tambahkan pernyataan CTAS ke kode SQL pekerjaan, dan restart dari titik simpan. Setelah tabel baru ditangkap, data akan direplikasi.

Batasan:

• Deteksi tabel baru didukung untuk VVR 8.0.1 atau versi yang lebih baru.

• Saat data disinkronkan dari tabel sumber CDC, hanya pekerjaan yang dimulai dalam mode initial yang dapat mendeteksi tabel baru.

• Konfigurasi tabel sumber yang ditambahkan menggunakan pernyataan CTAS baru harus sama dengan konfigurasi tabel sumber asli. Dengan demikian, vertex sumber dapat digunakan kembali.

• Parameter konfigurasi pekerjaan sebelum dan sesudah penambahan pernyataan CTAS harus sama. Misalnya, mode startup harus sama.

Prosedur:

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

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

3. Pada draf SQL pekerjaan, aktifkan deteksi tabel baru dan tambahkan pernyataan CTAS.

   1. Tambahkan pernyataan berikut untuk mengaktifkan deteksi tabel baru.

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

   2. Tambahkan pernyataan CTAS. Kode lengkap pekerjaan terlihat seperti berikut:

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

   3. Klik Deploy.

4. Pulihkan pekerjaan dari titik simpan.

   1. Pada halaman Deployments, klik nama deployment Anda.

   2. Pada halaman detail deployment, klik tab State. Lalu, klik subtab History.

   3. Pada daftar Savepoints, temukan titik simpan yang dibuat saat pekerjaan dibatalkan.

   4. Pilih More > Start job from this savepoint di kolom Actions. Untuk informasi selengkapnya, lihat Memulai penerapan Pekerjaan.

Menyinkronkan ke tabel partisi di Hologres

Deskripsi skenario: Mereplikasi data dari MySQL ke tabel partisi Hologres.

Catatan penggunaan: Jika primary key didefinisikan untuk tabel Hologres, kolom partisi harus disertakan dalam primary key.

Kode contoh:

Buat tabel MySQL:

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)
);

Bergantung pada apakah kolom partisi termasuk dalam primary key, pilih metode yang tepat:

• Jika primary key sumber berisi kolom partisi:

  • Gunakan pernyataan CTAS secara langsung.

    Hologres akan secara otomatis memverifikasi apakah kolom partisi disertakan dalam 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 primary key tabel sink dalam pernyataan CTAS dan sertakan kolom partisi dalam definisi primary key.

    Dalam kasus ini, tidak mendeklarasikan ulang primary key atau tidak menyertakan kolom partisi di dalamnya akan menyebabkan pekerjaan gagal.

    -- Deklarasikan bidang order_id, product_id, dan city sebagai primary key tabel partisi Hologres.
    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`;

Melebarkan tipe data selama replikasi

Deskripsi skenario: Selama sinkronisasi data, ubah presisi kolom, misalnya dari VARCHAR(10) ke VARCHAR(20), atau ubah tipe data kolom, seperti dari SMALLINT ke INT.

Metode:

• Pekerjaan baru: Aktifkan mode normalisasi tipe saat peluncuran pertama.

• Pekerjaan yang sudah ada: Hapus tabel sink Hologres, lalu restart tanpa state untuk menerapkan normalisasi tipe.

Aturan normalisasi tipe:

Jika tipe data baru dan asli dinormalisasi ke tipe data yang sama, perubahan tipe data berhasil dan pekerjaan berjalan normal. Jika tidak, pengecualian akan dilaporkan. Rinciannya sebagai berikut:

• TINYINT, SMALLINT, INT, dan BIGINT dikonversi menjadi BIGINT.

• CHAR, VARCHAR, dan STRING dikonversi menjadi STRING.

• FLOAT dan DOUBLE dikonversi menjadi DOUBLE.

• Tipe data lain dikonversi berdasarkan pemetaan tipe data antara bidang Hologres dan Flink. Untuk informasi selengkapnya, lihat Pemetaan tipe data.

Kode contoh:

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

Menyinkronkan data dari MongoDB ke Hologres

Batasan:

• Dukungan terbatas pada VVR 8.0.6 atau versi yang lebih baru dan MongoDB versi 6.0 atau lebih baru.

• Pada opsi konektor untuk tabel sumber, scan.incremental.snapshot.enabled dan scan.full-changelog harus diatur ke true.

• Fitur preimage dan postimage harus diaktifkan untuk database MongoDB. Untuk informasi selengkapnya, lihat Document Preimages.

• Untuk menyinkronkan data dari beberapa koleksi MongoDB dalam satu pekerjaan, pastikan konfigurasi opsi konektor berikut identik untuk semua tabel:

  • Opsi terkait database MongoDB, termasuk hosts, scheme, username, password, dan connectionOptions

  • scan.startup.mode

Kode contoh:

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;