Realtime Compute for Apache Flink:Stream storage Fluss

Visão geral

As tabelas de chave primária do Fluss suportam consumo completo e incremental. O sistema lê primeiro um snapshot histórico completo da tabela. Após a conclusão da leitura do snapshot, ele alterna automática e perfeitamente para o consumo de fluxo incremental. Esse processo garante a integridade e consistência dos dados.

Modos de consumo

Controle o ponto inicial de consumo usando o parâmetro scan.startup.mode.

Exemplos de SQL

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

Assinatura de log

As tabelas de chave primária do Fluss suportam nativamente a assinatura de log CDC (Change Data Capture). Isso permite que jobs Flink downstream consumam um fluxo completo de changelog (eventos INSERT, UPDATE_BEFORE, UPDATE_AFTER e DELETE) em tempo real, sem necessidade de configuração separada de binlog.

Isso significa que:

• Todas as alterações em uma tabela de chave primária geram logs CDC automaticamente.

• Aplicações downstream podem assinar diretamente a tabela de chave primária para receber o fluxo completo de changelog, destinado à sincronização de dados em tempo real ou ETL streaming.

Exemplo de SQL

-- 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.

Tabelas virtuais

Além de assinar o fluxo CDC diretamente, o Fluss fornece um mecanismo de tabela virtual. Você pode acessar changelogs com metadados adicionando um sufixo específico ao nome da tabela base, eliminando a necessidade de armazenamento extra de dados. O sistema gera essas tabelas virtuais automaticamente, dispensando criação manual.

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

Visão geral

O Fluss suporta atualizações parciais, permitindo que múltiplos fluxos de dados escrevam independentemente em suas respectivas colunas na mesma tabela ampla de chave primária. O motor do Fluss mescla automaticamente os campos no nível de registro com base na chave primária. Essa abordagem evita a necessidade de manter estados complexos de join de múltiplos fluxos no Flink e previne problemas relacionados a tamanhos elevados de estado.

Caso de uso

Um caso de uso típico é a construção de tabelas amplas de perfil de usuário em tempo real. Diferentes fontes de dados, como sistemas de recomendação, logs de impressão, logs de cliques e dados de carrinho de compras, podem atualizar seus respectivos campos na tabela de perfil do usuário. Esse processo cria uma visão 360 graus completa do usuário.

Limitações

• A tabela de destino deve ser uma tabela de chave primária.

• A operação de escrita de cada fluxo de dados deve incluir todos os campos da chave primária.

• A topologia de computação não deve conter nós de join de múltiplos fluxos. O Fluss lida com a operação de mesclagem diretamente na camada de armazenamento.

Exemplos de SQL

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

As três instruções INSERT acima podem ser executadas como jobs Flink separados. A camada de armazenamento do Fluss mescla automaticamente os campos para a mesma chave primária em linhas completas.

Visão geral

O Delta Join é um paradigma de join fluxo-fluxo de próxima geração fornecido pelo Fluss. Ele transfere as operações de atualização de dados para a camada de armazenamento do Fluss, aproveitando tabelas de chave primária e índices secundários para buscas eficientes. Este design elimina a necessidade de manter estado de join no Flink.

Principais vantagens:

• Sem estado de join: Elimina o armazenamento redundante de estado e evita problemas de garbage collection (GC) causados por estados grandes.

• Baixo consumo de recursos: Reduz o consumo de memória e CPU do Flink em mais de 86% nos testes.

• Maior estabilidade e eficiência: Evita gargalos de desempenho causados pelo inchaço do estado.

Comparação com joins tradicionais

Limitações

• Tanto a tabela da esquerda quanto a da direita devem ser tabelas de chave primária do Fluss (tabelas particionadas são suportadas).

• A bucket key deve ser um prefixo da chave primária.

• A chave de join deve incluir a bucket key. Para uma tabela particionada, deve incluir também a chave de partição.

Exemplos de SQL

1. Criar tabelas

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. Executar um Delta Join

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

3. Otimizar o desempenho de busca com hints

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;

Configuração de parâmetros do job

Para usar um Delta Join, adicione as seguintes configurações aos parâmetros de runtime do seu job Flink:

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'

Após ativar o Delta Join, o nó Delta Join aparece na visão geral do job Flink, indicando que a otimização está ativa.

Visão geral

Utilize uma tabela de chave primária do Fluss como uma tabela de dimensão no Flink SQL. Um lookup join permite enriquecer dados de streaming com dados dimensionais em tempo real. O Fluss suporta buscas pontuais KV eficientes e pesquisas por índice secundário, sendo adequado para cenários de join de tabela de dimensão com baixa latência.

Limitações

• Apenas uma tabela de chave primária pode ser usada como tabela de dimensão.

• A condição de join deve incluir todos os campos da chave primária da tabela de dimensão.

Modos síncrono e assíncrono

Exemplos de SQL

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

Visão geral

As tabelas de chave primária do Fluss permitem configurar um motor de mesclagem para controlar como o sistema combina registros com a mesma chave primária. Use o parâmetro table.merge-engine para selecionar uma das seguintes estratégias:

Exemplos de SQL

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

Visão geral

O Fluss suporta tabelas particionadas. Utilize PARTITIONED BY para dividir dados em diferentes partições físicas com base em uma chave de partição. Isso é ideal para gerenciar ciclos de vida de dados e melhorar o desempenho de consultas baseadas em tempo ou dimensões de negócio. Tabelas particionadas suportam particionamento de campo único e particionamento de múltiplos campos, oferecendo três estratégias ortogonais de gerenciamento de partição: gerenciamento manual, gerenciamento automático e criação dinâmica.

Regras básicas

• O tipo de dados de uma chave de partição deve ser STRING.

• Para uma tabela de chave primária, a chave de partição deve ser um subconjunto da chave primária.

• Configure regras de particionamento automático apenas durante a criação da tabela; não é possível modificá-las posteriormente.

Estratégias de gerenciamento de partição

Essas três estratégias são ortogonais e podem ser usadas em combinação. Por exemplo, ative o gerenciamento automático para criação e limpeza baseadas em tempo e a criação dinâmica como fallback para lidar com partições que não foram pré-criadas.

Exemplos de SQL

1. Criar uma tabela particionada por campo único com gerenciamento automático de partição

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. Criar uma tabela particionada por múltiplos campos

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

Nota : Para particionamento de múltiplos campos, especifique a chave baseada em tempo para particionamento automático usando table.auto-partition.key . O particionamento de múltiplos campos suporta apenas expiração e limpeza automáticas, não a pré-criação automática. O parâmetro num-precreate é forçado a 0.

3. Gerenciar partições manualmente

-- 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. Criar partições dinamicamente

Ao gravar dados, se a partição de destino não existir, o cliente a cria automaticamente. Este recurso está ativado por padrão.

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

O parâmetro do cliente client.writer.dynamic-create-partition.enabled (padrão: true) controla a criação dinâmica. Ele está sujeito aos limites de nível de cluster de max.partition.num (padrão: 1000) e max.bucket.num (padrão: 128000).

Pruning de partição

Ao consultar uma tabela particionada no modo streaming ou batch, o Fluss suporta pruning de partição. Ao filtrar pela chave de partição em uma cláusula WHERE, a consulta lê apenas as partições correspondentes, reduzindo a sobrecarga desnecessária de I/O.

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

Parâmetros de particionamento automático

Melhores práticas

• Escolha a granularidade de partição correta: Selecione uma granularidade de tempo com base no volume de dados e nos padrões de consulta. Para grandes volumes de dados e faixas de tempo de consulta estreitas, recomendamos particionar por DAY ou HOUR. Para volumes menores, particione por MONTH ou QUARTER.

• Defina uma política de retenção razoável: Use num-retention para controlar o número de partições históricas. Isso ajuda a evitar o inchaço de metadados e reduz a carga de gerenciamento do cluster causada por um número excessivo de partições.

• Combine gerenciamento automático e criação dinâmica: Utilize o gerenciamento automático para criação e limpeza rotineiras de partições baseadas em tempo e a criação dinâmica como fallback para lidar com dados anormais ou atrasados.

• Esteja ciente dos limites do cluster: O parâmetro max.partition.num (padrão: 1000) limita o número de partições por tabela. O número total de buckets no cluster é limitado por max.bucket.num (padrão: 128.000). O número total de buckets em uma tabela (número de partições × bucket.num) não deve exceder esse limite.

O Fluss usa o formato de armazenamento colunar Apache Arrow para transmissão de logs. Isso suporta o pushdown de operações de filtro e projeção para a camada de armazenamento, reduzindo o volume de transferência de dados e podendo melhorar o desempenho da consulta em até 10 vezes.

Este recurso é especialmente eficaz nos seguintes cenários:

• Leitura de apenas um subconjunto de colunas de uma tabela ampla (pushdown de projeção).

• Filtragem de uma grande quantidade de dados com base em condições específicas (pushdown de predicado).

O formato de armazenamento de log é controlado pelo parâmetro table.log.format, cujo padrão é ARROW.

Visão geral

O Fluss suporta uma arquitetura unificada stream-lake que pode sincronizar automaticamente dados em tempo real do Fluss para um data lake, como o Apache Paimon. Isso alcança o armazenamento em camadas automático para dados quentes e frios, equilibrando a baixa latência das consultas em tempo real com os baixos custos de armazenamento de dados históricos.

Principais recursos:

• Grava automaticamente dados de streaming do Fluss no armazenamento lake Paimon.

• Organiza automaticamente dados quentes e frios em camadas. Dados quentes são retidos no armazenamento de fluxo Fluss, enquanto dados frios são movidos para o Paimon.

• Fornece uma interface de consulta Flink SQL unificada para acessar dados tanto no armazenamento de fluxo quanto no lake simultaneamente.

Parâmetros de configuração

Visão geral

O Fluss suporta exploração rápida de dados em tabelas no modo batch do Flink SQL, incluindo consultas LIMIT, buscas pontuais por chave primária e agregações COUNT(*). Visualize rapidamente os dados em uma tabela sem iniciar um job de streaming contínuo. Isso é útil para desenvolvimento, depuração, validação de dados e troubleshooting.

Consultas LIMIT

Execute consultas LIMIT em tabelas de chave primária e tabelas append-only para recuperar rapidamente uma pequena amostra de dados.

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

Buscas pontuais por chave primária

Para tabelas de chave primária, realize buscas pontuais precisas especificando a chave primária em uma cláusula WHERE. Isso aproveita o motor de armazenamento KV do Fluss para alcançar tempos de resposta no nível de milissegundos.

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

Agregações COUNT(*)

Execute uma consulta COUNT(*) em uma tabela de chave primária para obter rapidamente o número total de registros.

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