Mengelola tabel - Realtime Compute for Apache Flink - Alibaba Cloud Documentation Center

Topik ini menjelaskan cara membuat dan mengelola tabel Fluss menggunakan Flink SQL.

Batasan

Fitur ini hanya tersedia di Realtime Compute for Apache Flink VVR 11.2 dan versi setelahnya.

Buat tabel primary key

Masuk ke Konsol Realtime Compute for Apache Flink.
Pada kolom Actions ruang kerja target, klik Console.
Di panel navigasi sebelah kiri, pilih Data Development > Data Exploration > New Temporary Query Script.
Tulis dan jalankan pernyataan SQL.

CREATE TABLE `my-catalog`.`my_db`.`my_pk_table` (
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT,
  PRIMARY KEY (shop_id, user_id) NOT ENFORCED
) WITH (
  'bucket.num' = '4'
);

Catatan

Anda harus merencanakan jumlah bucket saat membuat tabel.

Meskipun Anda dapat menyesuaikan bucketing untuk tabel log, jumlah bucket untuk tabel primary key tidak dapat diubah setelah pembuatan. Oleh karena itu, tentukan jumlah bucket berdasarkan volume data Anda saat membuat tabel. Gunakan rumus berikut:

Jumlah bucket = Total volume data dalam satu partisi / Ukuran bucket target

Buat tabel log

CREATE TABLE `my-catalog`.`my_db`.`my_log_table` (
  order_id BIGINT,
  item_id BIGINT,
  amount INT,
  address STRING
) WITH (
  'bucket.num' = '8'
);

Buat tabel partisi

Fluss saat ini hanya mendukung tipe data STRING untuk kunci partisi.
Untuk tabel primary key yang dipartisi, kunci partisi (bidang dt pada contoh berikut) harus merupakan subset dari primary key.

Tabel primary key partisi

Kunci partisinya adalah dt, dan primary key-nya adalah [dt, shop_id, user_id].

CREATE TABLE `my-catalog`.`my_db`.`my_part_pk_table` (
  dt STRING,
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT,
  PRIMARY KEY (dt, shop_id, user_id) NOT ENFORCED
) PARTITIONED BY (dt) WITH (
  'bucket.num' = '4'
);

Tabel log partisi

Contoh berikut membuat tabel log partisi dengan dt sebagai kunci partisi.

CREATE TABLE `my-catalog`.`my_db`.`my_part_log_table` (
  dt STRING,
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT
) PARTITIONED BY (dt) WITH (
  'bucket.num' = '4'
);

Tambahkan partisi

ALTER TABLE my-catalog.my_db.my_part_pk_table 
ADD PARTITION (dt = '2025-03-05');

Buat tabel auto-partisi

Tabel auto-partisi memperluas tabel partisi dengan partisi otomatis. Fluss membuat partisi terlebih dahulu berdasarkan kebijakan auto-partisi Anda.

Tabel primary key auto-partisi

Kunci partisinya adalah dt, dan interval auto-partisi adalah hari.

CREATE TABLE `my-catalog`.`my_db`.`my_auto_part_pk_table` (
  dt STRING,
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT,
  PRIMARY KEY (dt, shop_id, user_id) NOT ENFORCED
) PARTITIONED BY (dt) WITH (
  'bucket.num' = '4'
  'table.auto-partition.enabled' = 'true',
  'table.auto-partition.time-unit' = 'day'
);

Tabel log auto-partisi

Kunci partisinya adalah dt, dan interval auto-partisi adalah hari.

CREATE TABLE `my-catalog`.`my_db`.`my_auto_part_log_table` (
  dt STRING,
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT
) PARTITIONED BY (dt) WITH (
  'bucket.num' = '4'
  'table.auto-partition.enabled' = 'true',
  'table.auto-partition.time-unit' = 'day'
);

Perbarui tabel

Tambahkan kolom

Catatan

Fitur ini memerlukan Realtime Compute for Apache Flink VVR 11.5 atau versi setelahnya dan Fluss 0.8-ali-3.0 atau versi setelahnya. Untuk menghindari gangguan pada operasi baca dan tulis, pertama-tama upgrade pekerjaan Flink Anda ke VVR 11.5 atau versi setelahnya, lalu upgrade kluster Fluss Anda ke 0.8-ali-3.0.

Fluss memungkinkan Anda mengembangkan skema tabel dengan menambahkan kolom baru. Anda dapat menambahkan kolom dengan tipe data apa pun, termasuk tipe kompleks seperti ROW, MAP, dan ARRAY.

Ini adalah operasi ringan yang hanya melibatkan metadata dengan manfaat berikut:

Tidak ada penulisan ulang data: Anda tidak perlu menulis ulang atau memigrasikan file data yang sudah ada saat menambahkan kolom.
Eksekusi instan: Operasi selesai dalam hitungan milidetik, terlepas dari ukuran tabel.
Ketersediaan tinggi: Tabel tetap online dan sepenuhnya tersedia selama evolusi skema, tanpa gangguan pada klien yang sedang berjalan.

Saat ini fitur ini memiliki batasan berikut:

Urutan kolom: Kolom baru selalu ditambahkan di akhir daftar kolom yang sudah ada.
Nullability: Hanya kolom nullable yang dapat ditambahkan ke tabel untuk memastikan kompatibilitas dengan data yang sudah ada.
Bidang bersarang: Menambahkan bidang ke tipe ROW bersarang yang sudah ada saat ini belum didukung. Operasi jenis ini termasuk dalam kategori "memperbarui tipe kolom" dan akan didukung di rilis mendatang.

Gunakan pernyataan ALTER TABLE untuk menambahkan satu atau beberapa kolom.

-- Tambahkan satu kolom di akhir tabel
ALTER TABLE `my-catalog`.`my_db`.`my_pk_table` ADD user_email STRING COMMENT 'User email address';

-- Tambahkan beberapa kolom di akhir tabel
ALTER TABLE `my-catalog`.`my_db`.`my_pk_table` ADD (
    user_email STRING COMMENT 'User email address',
    order_quantity INT
);

Ubah properti tabel

Gunakan pernyataan ALTER TABLE SET untuk mengubah satu atau beberapa opsi konektor dan opsi penyimpanan. Nilai baru akan menimpa nilai yang sudah ada.

Kluster Fluss menerapkan perubahan opsi penyimpanan secara dinamis. Anda tidak perlu membuat ulang tabel.

Opsi yang didukung

Semua Opsi Baca, Opsi Tulis, Opsi Lookup, dan Opsi Lainnya, kecuali bootstrap.servers.
Opsi Penyimpanan berikut:
- table.datalake.enabled: Mengaktifkan atau menonaktifkan integrasi data lake dan stream.
- table.datalake.freshness: Menetapkan freshness data untuk penyimpanan data lake.

-- Aktifkan integrasi data lake dan stream
ALTER TABLE `my-catalog`.`my_db`.`my_pk_table` SET ('table.datalake.enabled' = 'true');

-- Tetapkan freshness data lake menjadi 5 menit
ALTER TABLE `my-catalog`.`my_db`.`my_pk_table` SET ('table.datalake.freshness' = '5min');

Penting

Setelah integrasi data lake dan stream (table.datalake.enabled) diaktifkan pada sebuah tabel, opsi awalan format lake seperti paimon.* tidak dapat lagi dimodifikasi.

Atur ulang properti tabel

Gunakan pernyataan ALTER TABLE RESET untuk mengembalikan satu atau beberapa opsi konektor dan opsi penyimpanan ke nilai default-nya.

-- Atur ulang integrasi data lake dan stream ke nilai default-nya (dinonaktifkan)
ALTER TABLE `my-catalog`.`my_db`.`my_pk_table` RESET ('table.datalake.enabled');

Klon tabel (kloning skema)

Sintaksis CREATE TABLE LIKE membuat tabel baru dengan skema, informasi partisi, dan properti yang sama seperti tabel yang sudah ada.

Anda harus menjalankan operasi ini dalam skrip kueri temporary. Untuk membuka skrip baru, pilih Data Development > Data Exploration > New Temporary Query Script.

-- Buat tabel temporary bernama datagen
CREATE TEMPORARY TABLE datagen (
 user_id BIGINT,
 item_id BIGINT,
 behavior STRING,
 dt STRING,
 hh STRING
) WITH (
 'connector' = 'datagen',
 'rows-per-second' = '10'
);
-- Buat tabel Fluss berdasarkan tabel datagen, tanpa menyertakan opsinya
CREATE TABLE my-catalog.my_db.my_like_db LIKE datagen (EXCLUDING OPTIONS);

Hapus tabel

DROP TABLE `my-catalog`.`my_db`.`my_pk_table`;