Apache Fluss Quickstart: Análises em Tempo Real com Flink

O Apache Fluss acaba de se formar no incubador da Apache — agora é um projeto de nível superior (top-level), aprovado por unanimidade pelo board. Se você está construindo aplicações agênticas, este vale o seu tempo: agentes precisam de contexto com atualização de menos de um segundo para tomar decisões, e um Lakehouse clássico (Paimon, Iceberg, Hudi) é ótimo para armazenar histórico, mas péssimo para responder “o que aconteceu dois segundos atrás”. O Fluss fecha essa lacuna com uma camada de armazenamento de streaming nativa para lakehouse.

Neste guia você executará o quickstart oficial de ponta a ponta: subir um cluster Fluss + Flink com Docker Compose, fazer streaming de dados, enriquecê-los com lookup joins e rodar análises em tempo real — tudo em Flink SQL. Sem necessidade de Java.

Por que Fluss, em um parágrafo

Fluss (alemão para “rio”) fica em cima do seu data lake, não no lugar dele. Dados quentes, atualizados com frequência, vivem no Fluss; o histórico de cauda longa permanece no Paimon/Iceberg/Hudi. Os motores leem uma visão unificada via Union Read, então os agentes obtêm contexto tanto fresco quanto completo. Por baixo dos panos, é streaming colunar sobre Apache Arrow, com poda de colunas e predicate pushdown no lado do servidor — os motores só leem os bytes de que realmente precisam.

As funcionalidades que importam para workloads de agentes:

• Tabelas com chave primária — point lookups de alto QPS, dedup, atualizações parciais, delta joins
• Changelogs — histórico append-only de mudanças de estado/decisões, útil para auditoria e reprodutibilidade
• Frescor de menos de um segundo — os dados ficam consultáveis assim que chegam
• Separação computação-armazenamento — o Fluss gerencia estado e armazenamento; Flink/Spark só calculam

Passo 1: Suba o cluster

Você precisa do Docker e do plugin Compose v2. Crie um diretório de trabalho com este docker-compose.yml (reduzido ao essencial):

services:
  # S3-compatible storage for tiered data
  rustfs:
    image: rustfs/rustfs:1.0.0-alpha.83
    ports: ["9000:9000", "9001:9001"]
    environment:
      - RUSTFS_ACCESS_KEY=rustfsadmin
      - RUSTFS_SECRET_KEY=rustfsadmin

  # Fluss cluster
  coordinator-server:
    image: apache/fluss:0.9.1-incubating
    command: coordinatorServer
    environment:
      - FLUSS_PROPERTIES=
          zookeeper.address: zookeeper:2181
          bind.listeners: FLUSS://coordinator-server:9123
          remote.data.dir: s3://fluss/remote-data
          s3.endpoint: http://rustfs:9000
          s3.access-key: rustfsadmin
          s3.secret-key: rustfsadmin

  tablet-server:
    image: apache/fluss:0.9.1-incubating
    command: tabletServer
    environment:
      - FLUSS_PROPERTIES=
          zookeeper.address: zookeeper:2181
          data.dir: /tmp/fluss/data
          remote.data.dir: s3://fluss/remote-data
          s3.endpoint: http://rustfs:9000
          s3.access-key: rustfsadmin
          s3.secret-key: rustfsadmin

  zookeeper:
    image: zookeeper:3.9.2

  # Flink cluster (bundles the Fluss connector + flink-faker)
  jobmanager:
    image: apache/fluss-quickstart-flink:1.20-0.9.1-incubating
    ports: ["8083:8081"]
    command: jobmanager

  taskmanager:
    image: apache/fluss-quickstart-flink:1.20-0.9.1-incubating
    command: taskmanager

  sql-client:
    image: apache/fluss-quickstart-flink:1.20-0.9.1-incubating
    command: /opt/sql-client/sql-client

Inicie tudo e verifique:

docker compose up -d
docker compose ps

A UI do Flink está em http://localhost:8083. O console do RustFS (seu bucket S3, fluss) está em http://localhost:9001, com as credenciais rustfsadmin/rustfsadmin.

Passo 2: Crie o catálogo Fluss e as tabelas

Entre no SQL client e aponte-o para o Fluss:

docker compose run sql-client

CREATE CATALOG fluss_catalog WITH (
  'type' = 'fluss',
  'bootstrap.servers' = 'coordinator-server:9123'
);
USE CATALOG fluss_catalog;

Crie tabelas com chave primária — elas viabilizam os point lookups rápidos:

CREATE TABLE fluss_order (
  `order_key`   BIGINT,
  `cust_key`    INT NOT NULL,
  `total_price` DECIMAL(15, 2),
  `order_date`  DATE,
  `order_priority` STRING,
  `ptime` AS PROCTIME(),
  PRIMARY KEY (`order_key`) NOT ENFORCED
);

CREATE TABLE fluss_customer (
  `cust_key`  INT NOT NULL,
  `name`      STRING,
  `phone`     STRING,
  `nation_key` INT NOT NULL,
  `acctbal`   DECIMAL(15, 2),
  `mktsegment` STRING,
  PRIMARY KEY (`cust_key`) NOT ENFORCED
);

Passo 3: Faça streaming dos dados

A imagem do quickstart já cria previamente tabelas de origem faker que geram dados de demonstração. Sincronize-as no Fluss:

EXECUTE STATEMENT SET
BEGIN
  INSERT INTO fluss_nation   SELECT * FROM `default_catalog`.`default_database`.source_nation;
  INSERT INTO fluss_customer SELECT * FROM `default_catalog`.`default_database`.source_customer;
  INSERT INTO fluss_order    SELECT * FROM `default_catalog`.`default_database`.source_order;
END;

Passo 4: Enriqueça com lookup joins

É aqui que o Fluss brilha: fazer join de um stream de pedidos contra tabelas com PK para busca de dimensões é uma operação de alto QPS, não um scan.

INSERT INTO enriched_orders
SELECT o.order_key, o.cust_key, o.total_price, o.order_date, o.order_priority,
       c.name, c.phone, c.acctbal, c.mktsegment, n.name
FROM fluss_order o
LEFT JOIN fluss_customer FOR SYSTEM_TIME AS OF `o`.`ptime` AS c ON o.cust_key = c.cust_key
LEFT JOIN fluss_nation   FOR SYSTEM_TIME AS OF `o`.`ptime` AS n ON c.nation_key = n.nation_key;

Passo 5: Análises em tempo real

SET 'sql-client.execution.result-mode' = 'tableau';
SET 'execution.runtime-mode' = 'batch';
SET 'table.dml-sync' = 'true';

SELECT * FROM enriched_orders LIMIT 2;

-- COUNT(*) is fast: Fluss maintains table-level stats, no full scan
SELECT COUNT(*) FROM enriched_orders;

-- point lookup by primary key
SELECT * FROM fluss_customer WHERE `cust_key` = 1;

Rode COUNT(*) algumas vezes — o número cresce conforme o faker continua produzindo, porque o Fluss ingere dados continuamente. Execute novamente e o resultado se atualiza em tempo real.

Passo 6: Atualização e exclusão

UPDATE fluss_customer SET `name` = 'fluss_updated' WHERE `cust_key` = 1;
SELECT * FROM fluss_customer WHERE `cust_key` = 1;  -- name is now fluss_updated

DELETE FROM fluss_customer WHERE `cust_key` = 1;
SELECT * FROM fluss_customer WHERE `cust_key` = 1;  -- empty

Pronto. Limpe com quit e depois docker compose down -v.

Notas de prática

• Não use as credenciais da demo em produção. O arquivo compose usa rustfsadmin com AssumeRole STS para o RustFS local. Em produção, use cadeias de credential providers específicas da nuvem.
• Use-o como armazenamento de contexto em tempo real de um agente. O Fluss mantém dedup, atualizações parciais e delta joins nativamente — o padrão típico de “real-time feature store” mapeia diretamente para uma tabela com PK.
• Changelogs são logs de decisão gratuitos. Para sistemas agênticos você quer uma trilha de auditoria das mudanças de estado; o Fluss gera changelogs append-only de fábrica.
• Ecossistema hoje: conectores Flink + Spark estão disponíveis, StarRocks está a caminho. Alibaba, Xiaohongshu, JD e Ant o utilizam em produção com tráfego em escala de bilhões.

Recursos

• GitHub: github.com/apache/fluss
• Documentação: fluss.apache.org/docs
• Este quickstart: fluss.apache.org/docs/quickstart/flink