Realtime Compute for Apache Flink:Akış depolama Fluss

Genel bakış

Fluss birincil anahtar tabloları tam ve artımlı tüketimi destekler. Sistem öncelikle tablonun tam bir geçmiş anlık görüntüsünü okur. Anlık görüntü okuması tamamlandıktan sonra, otomatik ve kesintisiz bir şekilde artımlı akış tüketimine geçer. Bu süreç, veri bütünlüğünü ve tutarlılığını sağlar.

Tüketim modları

scan.startup.mode parametresini kullanarak tüketim başlangıç noktasını kontrol edebilirsiniz.

SQL örnekleri

-- Full and incremental consumption (default mode)
SELECT * FROM `fluss_catalog`.`fluss_db`.orders;

-- Consume from a specific timestamp by using a hint
SELECT * FROM `fluss_catalog`.`fluss_db`.orders
  /*+ OPTIONS('scan.startup.mode' = 'timestamp', 'scan.startup.timestamp' = '2025-01-01 00:00:00') */;

-- Consume from the latest offset
SELECT * FROM `fluss_catalog`.`fluss_db`.orders
  /*+ OPTIONS('scan.startup.mode' = 'latest') */;

Günlük aboneliği

Fluss birincil anahtar tabloları yerel olarak CDC (Change Data Capture) günlük aboneliğini destekler. Bu, alt akıştaki Flink işlerinin ayrı bir binlog yapılandırması gerektirmeden eksiksiz bir değişiklik günlüğü akışını (INSERT, UPDATE_BEFORE, UPDATE_AFTER ve DELETE olayları) gerçek zamanlı olarak tüketmesine olanak tanır.

Bu şu anlama gelir:

• Bir birincil anahtar tablosundaki tüm değişiklikler otomatik olarak CDC günlükleri oluşturur.

• Alt akış uygulamaları, gerçek zamanlı veri senkronizasyonu veya akış ETL için eksiksiz değişiklik günlüğü akışını almak üzere doğrudan birincil anahtar tablosuna abone olabilir.

SQL örneği

-- Read the primary key table directly to get the CDC changelog stream
SELECT * FROM `fluss_catalog`.`fluss_db`.orders;
-- The output includes complete INSERT, UPDATE, and DELETE events.

Sanal tablolar

CDC akışına doğrudan abone olmanın yanı sıra, Fluss bir sanal tablo mekanizması sağlar. Ekstra veri depolamaya gerek kalmadan meta verilerle birlikte değişiklik günlüklerine erişmek için temel tablo adına belirli bir sonek ekleyebilirsiniz. Sistem bu sanal tabloları otomatik olarak oluşturur, manuel oluşturmaya gerek kalmaz.

-- Read the $changelog virtual table for the orders table
SELECT * FROM `fluss_catalog`.`fluss_db`.orders$changelog;

-- Consume from the latest offset by using a hint
SELECT * FROM `fluss_catalog`.`fluss_db`.orders$changelog
  /*+ OPTIONS('scan.startup.mode' = 'latest') */;

-- Filter for delete events
SELECT * FROM `fluss_catalog`.`fluss_db`.orders$changelog
WHERE _change_type = 'delete';

-- Read the $binlog virtual table for the users table
SELECT * FROM `fluss_catalog`.`fluss_db`.users$binlog;

-- Get the before and after values of a field from an update event
SELECT
  `before`.user_name AS old_name,
  `after`.user_name  AS new_name,
  _commit_timestamp  AS change_time
FROM `fluss_catalog`.`fluss_db`.users$binlog
WHERE _change_type = 'update';

Genel bakış

Fluss, birden çok veri akışının aynı geniş birincil anahtar tablosundaki kendi sütunlarına bağımsız olarak yazmasına olanak tanıyan kısmi güncellemeleri destekler. Fluss motoru, birincil anahtara dayalı olarak alanları kayıt düzeyinde otomatik olarak birleştirir. Bu yaklaşım, Flink'te karmaşık çoklu akış birleşim durumunu sürdürme ihtiyacını ortadan kaldırır ve büyük durum boyutlarıyla ilgili sorunları önler.

Kullanım durumu

Tipik bir kullanım durumu, gerçek zamanlı kullanıcı profili geniş tabloları oluşturmaktır. Öneri sistemleri, gösterim günlükleri, tıklama günlükleri ve alışveriş sepeti verileri gibi farklı veri kaynakları, kullanıcı profili tablosundaki kendi alanlarını güncelleyebilir. Bu süreç, kullanıcının eksiksiz bir 360 derecelik görünümünü oluşturur.

Sınırlamalar

• Hedef tablo bir birincil anahtar tablosu olmalıdır.

• Her veri akışının yazma işlemi, tüm birincil anahtar alanlarını içermelidir.

• Hesaplama topolojisi çoklu akış birleşim düğümleri içermemelidir. Fluss, birleştirme işlemini doğrudan depolama katmanında gerçekleştirir.

SQL örnekleri

-- 1. Create a wide table
CREATE TABLE IF NOT EXISTS `fluss_catalog`.fluss_db.user_rec_wide (
    user_id   STRING,
    item_id   STRING,
    rec_score DOUBLE,
    imp_cnt   INT,
    click_cnt INT,
    PRIMARY KEY (user_id, item_id) NOT ENFORCED
) WITH ('bucket.num' = '3');

-- 2. Use multiple streams to write to their respective fields
-- Stream 1: Writes recommendation scores
INSERT INTO `fluss_catalog`.fluss_db.user_rec_wide (user_id, item_id, rec_score)
SELECT user_id, item_id, rec_score 
FROM `fluss_catalog`.fluss_db.recommendations;

-- Stream 2: Writes impression counts
INSERT INTO `fluss_catalog`.fluss_db.user_rec_wide (user_id, item_id, imp_cnt)
SELECT user_id, item_id, imp_cnt 
FROM `fluss_catalog`.fluss_db.impressions;

-- Stream 3: Writes click counts
INSERT INTO `fluss_catalog`.fluss_db.user_rec_wide (user_id, item_id, click_cnt)
SELECT user_id, item_id, click_cnt 
FROM `fluss_catalog`.fluss_db.clicks;

Yukarıdaki üç INSERT ifadesi ayrı Flink işleri olarak çalıştırılabilir. Fluss depolama katmanı, aynı birincil anahtar için alanları otomatik olarak eksiksiz satırlarda birleştirir.

Genel bakış

Delta Join, Fluss tarafından sağlanan yeni nesil akış-akış birleşimi paradigmasıdır. Veri güncelleme işlemlerini Fluss depolama katmanına devrederek verimli aramalar için birincil anahtar tablolarından ve ikincil dizinlerden yararlanır. Bu tasarım, Flink'te birleşim durumunun sürdürülmesine olan ihtiyacı ortadan kaldırır.

Temel avantajlar:

• Birleşim durumu yok: Gereksiz durum depolamasını ortadan kaldırır ve büyük durum boyutlarından kaynaklanan garbage collection (GC) sorunlarını önler.

• Düşük kaynak tüketimi: Testlerde Flink bellek ve CPU tüketimini %86'nın üzerinde azaltır.

• Daha yüksek kararlılık ve verimlilik: Durum şişmesinden kaynaklanan performans darboğazlarını önler.

Geleneksel birleşimlerle karşılaştırma

Sınırlamalar

• Sol ve sağ tabloların her ikisi de Fluss birincil anahtar tablosu olmalıdır (bölümlenmiş tablolar desteklenir).

• Paket anahtarı, birincil anahtarın bir öneki olmalıdır.

• Birleşim anahtarı paket anahtarını içermelidir. Bölümlenmiş bir tablo için bölüm anahtarını da içermelidir.

SQL örnekleri

1. Tabloları oluşturun

CREATE TABLE orders (
  user_id BIGINT,
  order_date DATE,
  order_id BIGINT,
  amount DECIMAL(10, 2),
  PRIMARY KEY (user_id, order_id, order_date) NOT ENFORCED
) WITH ('bucket.key' = 'user_id, order_id');

CREATE TABLE order_enhance (
  user_id BIGINT,
  order_id BIGINT,
  risk_level TINYINT,
  is_fraud BOOLEAN,
  update_time TIMESTAMP(3),
  PRIMARY KEY (user_id, order_id) NOT ENFORCED
) WITH ('bucket.key' = 'user_id, order_id');

2. Delta Join çalıştırın

SELECT * FROM orders o
INNER JOIN order_enhance e
ON o.user_id = e.user_id AND o.order_id = e.order_id;

3. İpuçlarıyla arama performansını optimize edin

SELECT * FROM orders o /*+ OPTIONS('client.lookup.queue-size'='2560') */
INNER JOIN order_enhance e /*+ OPTIONS('client.lookup.queue-size'='2560') */
ON o.user_id = e.user_id AND o.order_id = e.order_id;

İş parametre yapılandırması

Delta Join kullanmak için Flink işinizin çalışma zamanı parametrelerine aşağıdaki yapılandırmaları ekleyin:

table.exec.delta-join.cache-enabled: 'true'
table.exec.async-lookup.buffer-capacity: '3000'
table.optimizer.delta-join.strategy: FORCE
table.exec.delta-join.left.cache-size: '100'
table.exec.delta-join.right.cache-size: '1000'

Delta Join'i etkinleştirdikten sonra, Flink işine genel bakışta Delta Join düğümü görünür; bu, optimizasyonun etkin olduğunu gösterir.

Genel bakış

Bir Fluss birincil anahtar tablosunu Flink SQL boyut tablosu olarak kullanabilirsiniz. Arama birleşimi, akış verilerini boyutsal verilerle gerçek zamanlı olarak zenginleştirmenize olanak tanır. Fluss, verimli KV nokta aramalarını ve ikincil dizin aramalarını destekleyerek düşük gecikmeli boyut tablosu birleştirme senaryoları için uygun hale gelir.

Sınırlamalar

• Boyut tablosu olarak yalnızca bir birincil anahtar tablosu kullanılabilir.

• Birleşim koşulu, boyut tablosunun tüm birincil anahtar alanlarını içermelidir.

Eşzamanlı ve eşzamansız modlar

SQL örnekleri

-- Configure lookup cache by using a hint to perform a dimension table join
SELECT
  o.order_id,
  o.amount,
  u.user_name,
  p.product_name
FROM `fluss_catalog`.`fluss_db`.orders AS o
LEFT JOIN `fluss_catalog`.`fluss_db`.user_dim
  /*+ OPTIONS('lookup.cache' = 'PARTIAL', 'lookup.partial-cache.expire-after-write' = '60s', 'lookup.partial-cache.max-rows' = '10000') */
  FOR SYSTEM_TIME AS OF o.proctime AS u
  ON o.user_id = u.user_id
LEFT JOIN `fluss_catalog`.`fluss_db`.product_dim
  FOR SYSTEM_TIME AS OF o.proctime AS p
  ON o.product_id = p.product_id;

-- Use synchronous lookup mode (suitable for debugging)
SELECT o.order_id, o.amount, d.user_name
FROM `fluss_catalog`.`fluss_db`.orders AS o
LEFT JOIN `fluss_catalog`.`fluss_db`.user_dim
  /*+ OPTIONS('lookup.async' = 'false') */
  FOR SYSTEM_TIME AS OF o.proctime AS d
  ON o.user_id = d.user_id;

Genel bakış

Fluss birincil anahtar tabloları, sistemin aynı birincil anahtara sahip kayıtları nasıl birleştireceğini kontrol etmek için bir birleştirme motoru yapılandırmanıza olanak tanır. Aşağıdaki stratejilerden birini seçmek için table.merge-engine parametresini kullanın:

SQL örnekleri

-- Use first_row for deduplication
CREATE TABLE IF NOT EXISTS `fluss_catalog`.fluss_db.event_dedup (
    event_id STRING,
    event_type STRING,
    event_time TIMESTAMP(3),
    payload STRING,
    PRIMARY KEY (event_id) NOT ENFORCED
) WITH (
  'table.merge-engine' = 'first_row'
);

-- Use versioned to update based on a version column
CREATE TABLE IF NOT EXISTS `fluss_catalog`.fluss_db.user_profile (
    user_id BIGINT,
    user_name STRING,
    age INT,
    update_version BIGINT,
    PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'table.merge-engine' = 'versioned',
  'table.merge-engine.versioned.ver-column' = 'update_version'
);

Genel bakış

Fluss bölümlenmiş tabloları destekler. Verileri bir bölüm anahtarına göre farklı fiziksel bölümlere ayırmak için PARTITIONED BY kullanabilirsiniz. Bu, zaman veya iş boyutlarına göre veri yaşam döngülerini yönetmek ve sorgu performansını artırmak için idealdir. Bölümlenmiş tablolar tek alanlı bölümlendirmeyi ve çok alanlı bölümlendirmeyi destekler ve üç dik bölümlendirme yönetimi stratejisi sunar: manuel yönetim, otomatik yönetim ve dinamik oluşturma.

Temel kurallar

• Bir bölüm anahtarının veri türü STRING olmalıdır.

• Bir birincil anahtar tablosu için bölüm anahtarı, birincil anahtarın bir alt kümesi olmalıdır.

• Otomatik bölümlendirme kurallarını yalnızca tabloyu oluştururken yapılandırabilirsiniz; bunları daha sonra değiştiremezsiniz.

Bölüm yönetimi stratejileri

Bu üç strateji diktir ve kombinasyon halinde kullanılabilir. Örneğin, zamana dayalı oluşturma ve temizleme için otomatik yönetimi ve önceden oluşturulmamış bölümleri işlemek için bir geri dönüş mekanizması olarak dinamik oluşturmayı aynı anda etkinleştirebilirsiniz.

SQL örnekleri

1. Otomatik bölüm yönetimi ile tek alanlı bölümlenmiş bir tablo oluşturun

CREATE TABLE `fluss_catalog`.fluss_db.site_access (
  site_id BIGINT,
  user_name STRING,
  pv BIGINT,
  event_day STRING,
  PRIMARY KEY (site_id, event_day) NOT ENFORCED
) PARTITIONED BY (event_day) WITH (
  'table.auto-partition.enabled' = 'true',
  'table.auto-partition.time-unit' = 'DAY',
  'table.auto-partition.num-precreate' = '5',
  'table.auto-partition.num-retention' = '30',
  'table.auto-partition.time-zone' = 'Asia/Shanghai'
);

2. Çok alanlı bölümlenmiş bir tablo oluşturun

CREATE TABLE `fluss_catalog`.fluss_db.region_access (
  access_id BIGINT,
  user_name STRING,
  event_date STRING,
  region STRING,
  PRIMARY KEY (access_id, event_date, region) NOT ENFORCED
) PARTITIONED BY (event_date, region) WITH (
  'table.auto-partition.enabled' = 'true',
  'table.auto-partition.key' = 'event_date',
  'table.auto-partition.time-unit' = 'DAY',
  'table.auto-partition.num-retention' = '7'
);

Not : Çok alanlı bölümlendirme için, otomatik bölümlendirme için zamana dayalı anahtarı table.auto-partition.key kullanarak belirtmeniz gerekir. Çok alanlı bölümlendirme yalnızca otomatik süre sonu ve temizlemeyi destekler, otomatik önceden oluşturmayı desteklemez. num-precreate parametresi 0'a zorlanır.

3. Bölümleri manuel olarak yönetin

-- View all partitions
SHOW PARTITIONS `fluss_catalog`.fluss_db.site_access;

-- Manually add a partition
ALTER TABLE `fluss_catalog`.fluss_db.site_access
  ADD PARTITION (event_day = '20250315');

-- Manually drop a partition
ALTER TABLE `fluss_catalog`.fluss_db.site_access
  DROP PARTITION (event_day = '20250101');

4. Bölümleri dinamik olarak oluşturun

Veri yazdığınızda, hedef bölüm yoksa istemci bunu otomatik olarak oluşturur. Bu özellik varsayılan olarak etkindir.

-- Write data. If the partition '20250701' does not exist, it is automatically created.
INSERT INTO `fluss_catalog`.fluss_db.site_access
VALUES (1, 'hello', 100, '20250701');

client.writer.dynamic-create-partition.enabled istemci parametresi (varsayılan: true) dinamik oluşturmayı kontrol eder. max.partition.num (varsayılan: 1000) ve max.bucket.num (varsayılan: 128000) küme düzeyindeki sınırlamalara tabidir.

Bölüm budama

Akış veya toplu iş modunda bölümlenmiş bir tabloyu sorguladığınızda, Fluss bölüm budamayı destekler. Bir WHERE yan tümcesinde bölüm anahtarına göre filtreleme yaparak, sorgu yalnızca eşleşen bölümleri okur; bu da gereksiz G/Ç yükünü azaltır.

-- Read only the data from the partition for a specific date
SELECT * FROM `fluss_catalog`.`fluss_db`.site_access
WHERE event_day = '20250315';

-- Filter a multi-field partitioned table by region
SELECT * FROM `fluss_catalog`.`fluss_db`.region_access
WHERE region = 'US';

Otomatik bölümleme parametreleri

En iyi uygulamalar

• Doğru bölüm hassasiyetini seçin: Veri hacminize ve sorgu desenlerinize uygun bir zaman hassasiyeti belirleyin. Büyük veri hacimleri ve dar sorgu zaman aralıkları için DAY veya HOUR ile bölümleme önerilir. Daha küçük veri hacimleri için MONTH veya QUARTER ile bölümleme yapabilirsiniz.

• Mantıklı bir saklama ilkesi belirleyin: Geçmiş bölüm sayısını kontrol etmek için num-retention kullanın. Bu yaklaşım, meta veri şişmesini önlemeye yardımcı olur ve aşırı sayıda bölümden kaynaklanan küme yönetim yükünü azaltır.

• Otomatik yönetim ve dinamik oluşturma yöntemlerini birleştirin: Rutin zaman tabanlı bölüm oluşturma ve temizleme işlemleri için otomatik yönetimi kullanın; anormal veya gecikmeli verileri işlemek için dinamik oluşturmayı yedek çözüm olarak uygulayın.

• Küme sınırlarını göz önünde bulundurun: max.partition.num parametresi (varsayılan: 1000), tablo başına düşen bölüm sayısını sınırlar. Kümedeki toplam paket sayısı max.bucket.num (varsayılan: 128.000) ile sınırlıdır. Bir tablodaki toplam paket sayısı (bölüm sayısı × bucket.num) bu sınırı aşmamalıdır.

Fluss, günlük iletimi için Apache Arrow sütunlu depolama biçimini kullanır. Bu yapı, filtreleme ve projeksiyon işlemlerinin depolama katmanına aktarılmasını destekler; böylece veri transfer hacmi azalır ve sorgu performansı 10 kata kadar artabilir.

Bu özellik özellikle aşağıdaki senaryolarda etkilidir:

• Geniş bir tablodan yalnızca belirli sütunların okunması (projeksiyon aşağı itme).

• Belirli koşullara dayalı olarak büyük miktardaki verinin filtrelenmesi (yüklem aşağı itme).

Günlük depolama biçimi, varsayılan değeri ARROW olan table.log.format parametresi ile kontrol edilir.

Genel Bakış

Fluss, gerçek zamanlı verileri Fluss'tan Apache Paimon gibi bir veri gölüne otomatik olarak senkronize edebilen birleşik bir akış-göl mimarisini destekler. Bu sayede sıcak ve soğuk veriler için otomatik katmanlı depolama sağlanır; gerçek zamanlı sorguların düşük gecikme süresi ile geçmiş verilerin düşük depolama maliyeti arasında denge kurulur.

Temel özellikler:

• Fluss'tan gelen akış verilerini otomatik olarak Paimon göl depolamasına yazar.

• Sıcak ve soğuk verileri otomatik olarak katmanlara ayırır. Sıcak veriler Fluss akış depolamasında tutulurken soğuk veriler Paimon'a taşınır.

• Hem akış hem de göl depolamasındaki verilere aynı anda erişmek için birleşik bir Flink SQL sorgu arabirimi sunar.

Yapılandırma parametreleri

Genel Bakış

Fluss, Flink SQL toplu iş modunda tablolar üzerinde hızlı veri keşfini destekler. Bu kapsamda LIMIT sorguları, birincil anahtar nokta aramaları ve COUNT(*) toplamları gerçekleştirilebilir. Sürekli çalışan bir akış işi başlatmadan tablodaki verileri hızla görüntüleyebilirsiniz. Bu yetenek geliştirme, hata ayıklama, veri doğrulama ve sorun giderme süreçlerinde faydalıdır.

LIMIT sorguları

Birincil anahtarlı ve yalnızca ekleme yapılabilen tablolar üzerinde LIMIT sorguları çalıştırarak küçük bir veri örneğini hızlıca alabilirsiniz.

-- View the first 10 records from a primary key table
SELECT * FROM `fluss_catalog`.`fluss_db`.orders LIMIT 10;

-- View the first 5 records from an append-only table
SELECT * FROM `fluss_catalog`.`fluss_db`.events LIMIT 5;

-- Combine with projection to view only specific columns
SELECT order_id, amount, order_time 
FROM `fluss_catalog`.`fluss_db`.orders LIMIT 20;

Birincil anahtar nokta aramaları

Birincil anahtarlı tablolar için WHERE yan tümcesinde birincil anahtarı belirterek kesin nokta aramaları yapabilirsiniz. Bu yöntem, Fluss KV depolama motorundan yararlanarak milisaniye düzeyinde yanıt süreleri sağlar.

-- Perform a point lookup for a single record by its primary key
SELECT * FROM `fluss_catalog`.`fluss_db`.orders
WHERE order_id = 1001;

-- Perform a point lookup with a composite primary key
SELECT * FROM `fluss_catalog`.`fluss_db`.user_orders
WHERE user_id = 12345 AND order_id = 6789;

COUNT(*) toplamları

Birincil anahtarlı bir tablo üzerinde COUNT(*) sorgusu çalıştırarak toplam kayıt sayısını hızlıca elde edebilirsiniz.

-- Count the total number of rows in a primary key table
SELECT COUNT(*) FROM `fluss_catalog`.`fluss_db`.orders;