Tabloları yönetin - Realtime Compute for Apache Flink - Alibaba Cloud Dokümantasyon Merkezi

Bu konu, Flink SQL kullanarak Fluss tablolarının nasıl oluşturulacağını ve yönetileceğini açıklar.

Sınırlamalar

Bu özellik yalnızca Realtime Compute for Apache Flink VVR 11.2 ve sonraki sürümlerde kullanılabilir.

Birincil anahtar tablosu oluşturma

Realtime Compute for Apache Flink konsolu üzerinde oturum açın.
Hedef çalışma alanının Actions sütununda Console seçeneğine tıklayın.
Sol gezinme bölmesinde Development > Scripts > New Temporary Script yolunu izleyin.
SQL ifadesini yazın ve çalıştırın.

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

Not

Tablo oluştururken paket sayısını planlamanız gerekir.

Günlük tablosu için paketleme ayarlarını değiştirebilmenize rağmen birincil anahtar tablosundaki paket sayısı oluşturulduktan sonra değiştirilemez. Bu nedenle tabloyu oluştururken veri hacminize göre paket sayısını belirlemelisiniz. Aşağıdaki formülü kullanın:

Paket sayısı = Tek bir bölümdeki toplam veri hacmi / Hedeflenen paket boyutu

Günlük tablosu oluşturma

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

Bölümlenmiş tablo oluşturma

Fluss şu anda yalnızca bölüm anahtarları için STRING veri türünü destekler.
Bölümlenmiş birincil anahtar tablosunda bölüm anahtarı (aşağıdaki örnekteki dt alanı) birincil anahtarın alt kümesi olmalıdır.

Bölümlenmiş birincil anahtar tablosu

Bölüm anahtarı dt, birincil anahtar ise [dt, shop_id, user_id] şeklindedir.

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

Bölümlenmiş günlük tablosu

Aşağıdaki örnekte dt bölüm anahtarı olarak kullanılarak bölümlenmiş bir günlük tablosu oluşturulur.

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

Bölüm ekleme

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

Otomatik bölümlenmiş tablo oluşturma

Otomatik bölümlenmiş tablo, otomatik bölümleme özelliğiyle genişletilmiş bir bölümlenmiş tablodur. Fluss, otomatik bölümleme ilkenize göre bölümleri önceden oluşturur.

Otomatik bölümlenmiş birincil anahtar tablosu

Bölüm anahtarı dt, otomatik bölümleme aralığı ise gündür.

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

Otomatik bölümlenmiş günlük tablosu

Bölüm anahtarı dt, otomatik bölümleme aralığı ise gündür.

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

Tabloyu güncelleme

Sütun ekleme

Not

Bu özellik için Realtime Compute for Apache Flink VVR 11.5 veya sonraki bir sürüm ile Fluss 0,8-ali-3,0 veya sonraki bir sürüm gereklidir. Okuma ve yazma işlemlerinin kesintiye uğramaması için önce Flink işinizi VVR 11.5 veya sonraki bir sürüme yükseltin, ardından Fluss kümenizi 0,8-ali-3,0 sürümüne yükseltin.

Fluss, yeni sütunlar ekleyerek tablonun şemasını geliştirmenize olanak tanır. ROW, MAP ve ARRAY gibi karmaşık türler dahil olmak üzere herhangi bir veri türünde sütun ekleyebilirsiniz.

Bu işlem yalnızca meta verileri etkileyen hafif bir işlemdir ve aşağıdaki avantajları sunar:

Sıfır veri yeniden yazma: Sütun eklerken mevcut veri dosyalarını yeniden yazmanız veya geçiş yapmanız gerekmez.
Anında yürütme: İşlem, tablo boyutundan bağımsız olarak milisaniyeler içinde tamamlanır.
Yüksek kullanılabilirlik: Şema gelişimi sırasında tablo çevrimiçi ve tamamen kullanılabilir durumda kalır; çalışan istemcilerde herhangi bir kesinti oluşmaz.

Bu özelliğin şu anda aşağıdaki sınırlamaları vardır:

Sütun sırası: Yeni sütunlar her zaman mevcut sütun listesinin sonuna eklenir.
Boş bırakılabilirlik: Mevcut verilerle uyumluluğu sağlamak için tabloya yalnızca boş bırakılabilir sütunlar eklenebilir.
İç içe geçmiş alanlar: Mevcut iç içe geçmiş ROW türüne alan eklenmesi şu anda desteklenmemektedir. Bu tür bir işlem "sütun türlerini güncelleme" kapsamında yer alır ve gelecekteki bir sürümde desteklenecektir.

Tek bir sütun veya birden çok sütun eklemek için ALTER TABLE ifadesini kullanın.

-- Add a single column at the end of the table
ALTER TABLE `my-catalog`.`my_db`.`my_pk_table` ADD user_email STRING COMMENT 'User email address';

-- Add multiple columns at the end of the table
ALTER TABLE `my-catalog`.`my_db`.`my_pk_table` ADD (
    user_email STRING COMMENT 'User email address',
    order_quantity INT
);

Tablo özelliklerini değiştirme

Bir veya daha fazla bağlayıcı seçeneğini ve depolama seçeneğini değiştirmek için ALTER TABLE SET ifadesini kullanın. Yeni değerler mevcut olanların üzerine yazılır.

Fluss kümesi depolama seçeneği değişikliklerini dinamik olarak uygular. Tabloyu yeniden oluşturmanız gerekmez.

Desteklenen seçenekler

bootstrap.servers hariç tüm Okuma Seçenekleri, Yazma Seçenekleri, Arama Seçenekleri ve Diğer Seçenekler.
Aşağıdaki Depolama Seçenekleri:
- table.datalake.enabled: Veri gölü ve akış entegrasyonunu etkinleştirir veya devre dışı bırakır.
- table.datalake.freshness: Veri gölü depolaması için veri tazelik süresini ayarlar.

-- Enable data lake and stream integration
ALTER TABLE `my-catalog`.`my_db`.`my_pk_table` SET ('table.datalake.enabled' = 'true');

-- Set data lake freshness to 5 minutes
ALTER TABLE `my-catalog`.`my_db`.`my_pk_table` SET ('table.datalake.freshness' = '5min');

Önemli

Bir tabloda veri gölü ve akış entegrasyonu (table.datalake.enabled) etkinleştirildikten sonra paimon.* gibi göl biçimi ön ek seçenekleri artık değiştirilemez.

Tablo özelliklerini sıfırlama

Bir veya daha fazla bağlayıcı seçeneğini ve depolama seçeneğini varsayılan değerlerine geri yüklemek için ALTER TABLE RESET ifadesini kullanın.

-- Reset data lake and stream integration to its default (disabled)
ALTER TABLE `my-catalog`.`my_db`.`my_pk_table` RESET ('table.datalake.enabled');

Tabloyu klonlama (şema klonlama)

CREATE TABLE LIKE sözdizimi, mevcut bir tabloyla aynı şemaya, bölüm bilgilerine ve özelliklere sahip yeni bir tablo oluşturur.

Bu işlemi geçici bir sorgu betiğinde çalıştırmanız gerekir. Yeni bir betik açmak için Development > Scripts > New Temporary Script yolunu izleyin.

-- Create a temporary table named datagen
CREATE TEMPORARY TABLE datagen (
 user_id BIGINT,
 item_id BIGINT,
 behavior STRING,
 dt STRING,
 hh STRING
) WITH (
 'connector' = 'datagen',
 'rows-per-second' = '10'
);
-- Create a Fluss table based on the datagen table, excluding its options
CREATE TABLE my-catalog.my_db.my_like_db LIKE datagen (EXCLUDING OPTIONS);

Tabloyu silme

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