Apache Paimon 实战：流式湖仓一体与实时数仓落地

当 Kafka 里的订单还在以每秒上万条的速率滚动，业务却想知道「实时 GMV 和昨天的对账差多少」——传统数仓要等 T+1，纯流计算又留不下历史。Apache Paimon 正是为解决这个矛盾而生：它把流式湖仓（Streaming Lakehouse）做成了一张「既能实时写、又能随时查」的表。本文用 Docker + Flink 快速上手 Paimon，演示 Kafka 实时入湖、主键表更新以及同一张表的离在线一体化查询，并给出与 Iceberg/Hudi 的选型建议。

一、Apache Paimon 是什么

Apache Paimon（原 Flink Table Store，已毕业为 Apache 顶级项目）是一种面向流批一体的湖仓存储格式。它构建在对象存储或 HDFS 之上，底层用 LSM（Log-Structured Merge-Tree）结构组织数据，因此既支持像消息流一样持续追加写入，也支持基于主键的更新、删除与聚合。你可以把它理解为「给数据湖加上了数据库的更新能力」：上游是 CDC 或 Kafka，下游同一个 SQL 既能跑流式最新值，也能跑批式全量分析。

相比本站的 Apache Iceberg 数据湖表格式 更偏向批式 ACID 与海量分析，Paimon 的差异化在于「为流而生」——写入即合并、读取可指定新鲜度，天然适配 Flink 的流批执行模式。两者并非互斥，很多架构里 Paimon 负责实时层、Iceberg 负责归档层。

二、核心概念先理清

• 主键表（Primary Key Table）：声明 PRIMARY KEY，支持更新/删除，适合订单、用户画像等需要「最新状态」的场景。底层按 bucket 分桶，写入时做去重与合并。
• 仅追加表（Append-only Table）：没有主键，只负责写入，适合日志、埋点等不可变数据。
• Bucket：数据分桶数，决定读写并行度与文件大小，主键表必须 bucket > 0。
• Changelog Producer：决定表能否产出变更流，取值 input / full / lookup / none，做流式下游时很关键。
• Merge Engine：主键冲突时的合并策略，如 deduplicate（保留最新行）、partial-update（字段级合并）、aggregation（聚合）。

三、用 Docker 起一套 Flink + Paimon 环境

最稳妥的方式是跑一个 Flink Standalone 集群，并把 Paimon 的 Flink 连接器 jar 放进 lib 目录。先准备 docker-compose：

version: "3"
services:
  jobmanager:
    image: flink:1.17.2
    command: jobmanager
    ports:
      - "8081:8081"
    volumes:
      - ./lib:/opt/flink/lib
  taskmanager:
    image: flink:1.17.2
    command: taskmanager
    depends_on:
      - jobmanager
    volumes:
      - ./lib:/opt/flink/lib
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
  kafka:
    image: confluentinc/cp-kafka:7.5.0
    depends_on: [zookeeper]
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

把与 Flink 大版本匹配的 Paimon 连接器放进去即可，例如 paimon-flink-1.17-0.9.0.jar（从 Maven Central 的 org/apache/paimon/paimon-flink-1.17 下载）。也可以不依赖 compose，直接下载 jar 后用 SQL Client 加载：

wget https://repo1.maven.org/maven2/org/apache/paimon/paimon-flink-1.17/0.9.0/paimon-flink-1.17-0.9.0.jar -O lib/paimon-flink.jar
docker compose up -d
docker compose exec jobmanager ./bin/sql-client.sh

四、用 Flink SQL 建库建表

进入 SQL Client 后，先注册 Paimon Catalog（warehouse 指向存储根路径），再建库建表。下面建一张订单主键表，按 order_id 去重、分 4 个桶，并开启 input 级 changelog：

CREATE CATALOG paimon_catalog WITH (
  'type' = 'paimon',
  'warehouse' = 'file:/tmp/paimon/warehouse'
);
USE CATALOG paimon_catalog;
CREATE DATABASE IF NOT EXISTS default;

CREATE TABLE IF NOT EXISTS default.orders (
  order_id   BIGINT,
  user_id    BIGINT,
  amount     DECIMAL(10,2),
  status     STRING,
  ts         TIMESTAMP(3),
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
  'bucket' = '4',
  'changelog-producer' = 'input',
  'merge-engine' = 'deduplicate'
);

再把 Kafka 上的原始订单流声明成一张外部表，作为入湖的数据源：

CREATE TABLE default.kafka_orders (
  order_id   BIGINT,
  user_id    BIGINT,
  amount     DECIMAL(10,2),
  status     STRING,
  ts         TIMESTAMP(3)
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders',
  'properties.bootstrap.servers' = 'kafka:9092',
  'properties.group.id' = 'paimon-ingest',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
);

五、Kafka 实时入湖

实时入湖就是一条 INSERT INTO … SELECT 的流式作业。Flink 会持续把 Kafka 里的订单写进 Paimon，并自动按主键做更新合并：

INSERT INTO default.orders
SELECT order_id, user_id, amount, status, ts
FROM default.kafka_orders;

如果该作业以流式（streaming）模式常驻，Paimon 表就始终持有「最新订单状态」。配合 Kafka 消息队列入门 的基础，这条链路把「消息流」无缝落成了「可查询的表」。若订单状态会变化（待支付→已支付→已退款），由于表声明了主键与 deduplicate 合并引擎，后到的同 order_id 记录会自然覆盖旧值。

六、流批一体查询：同一张表，两种读法

Paimon 的杀手锏是「一张表两种读法」。默认 streaming 模式读最新快照，batch 模式则做全量扫描——非常适合既要实时大屏、又要离线对账的场景：

-- 流式：拿最新状态（实时大屏）
SELECT status, COUNT(*) AS cnt, SUM(amount) AS gmv
FROM default.orders
GROUP BY status;

-- 批式：全量分析（对账 / 报表）
SET 'execution.runtime-mode' = 'batch';
SELECT user_id, SUM(amount) AS total
FROM default.orders
GROUP BY user_id
ORDER BY total DESC
LIMIT 10;

还支持时间旅行，按快照或时间戳回看历史，排障和对账非常方便：

-- 读指定快照
SELECT * FROM default.orders /*+ OPTIONS('scan.snapshot-id' = '1') */;

-- 读某个时间点之前的状态
SELECT * FROM default.orders
/*+ OPTIONS('scan.timestamp-millis' = '1727000000000') */;

六之一、主键更新语义：partial-update 字段级合并

上面的订单表用的是默认 deduplicate 合并引擎——同主键后到的整行直接覆盖前一行。但真实业务里，不同系统往往只更新用户画像的「不同列」：点击流负责打标签，交易流负责更新等级。这时用 partial-update 引擎，多条流写不同字段，Paimon 会把非空字段合并成同一行，而不是互相覆盖：

CREATE TABLE default.user_profile (
  user_id BIGINT,
  tags     STRING,
  level    INT,
  updated  TIMESTAMP(3),
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'bucket' = '4',
  'merge-engine' = 'partial-update',
  'changelog-producer' = 'input'
);

-- 点击流只写 tags
INSERT INTO default.user_profile (user_id, tags, updated)
SELECT user_id, collect_tags(tag), NOW() FROM click_stream GROUP BY user_id;

-- 交易流只写 level，不会清掉 tags
INSERT INTO default.user_profile (user_id, level, updated)
SELECT user_id, vip_level, NOW() FROM order_stream;

partial-update 的代价是同一主键的多列更新无法按「整行时间戳」判定先后，只在列级别合并。如果你的场景要求「后到整行胜出」，就退回 deduplicate；如果要求「某列取最大值」，还可以用 aggregation 引擎配合 sum/max 聚合函数。选错引擎是 Paimon 最常见的坑之一。

六之二、CDC 入湖：从 MySQL Binlog 实时同步

Kafka 入湖适合已有消息总线的团队；如果上游就是 MySQL 业务库，更直接的方式是用 Flink CDC 连接器读 Binlog，把 OLTP 数据「原样」同步进湖，免去了自己造 Kafka topic 的环节，且天然支持 exactly-once 与 schema 变更：

CREATE TABLE default.mysql_orders (
  order_id BIGINT,
  user_id  BIGINT,
  amount   DECIMAL(10,2),
  status   STRING,
  ts       TIMESTAMP(3),
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = 'mysql',
  'port' = '3306',
  'username' = 'cdc_user',
  'password' = 'secret',
  'database-name' = 'shop',
  'table-name' = 'orders'
);

INSERT INTO default.orders SELECT * FROM default.mysql_orders;

注意 mysql-cdc 需要额外的 flink-sql-connector-mysql-cdc jar 放进 lib。相比「业务写双份 / 定时 sqoop 全量抽取」，CDC 入湖把延迟从小时级压到秒级，且不会对业务库造成全表扫描压力——这也是 Paimon 被用来替换传统 T+1 数仓的关键一步。

六之三、小文件治理：compaction 调优

LSM 结构在高频写入下会不断产生小文件，读放大随之上升。Paimon 默认在后台自动 compaction，但高吞吐场景需要主动调大写入缓冲与触发阈值，避免「写得快、查得慢」：

ALTER TABLE default.orders SET OPTIONS (
  'write-buffer-size' = '256 MB',
  'num-sorted-run.compaction-trigger' = '5',
  'compact.min.file-num' = '5',
  'compact.max.file-num' = '50'
);

经验上：写入越猛，write-buffer-size 越大；读多写少则可以把 compaction 触发阈值调低以加速合并。compact.max.file-num 控制单次合并上限，过大会让单次 compaction 耗时过长、拖累实时性。生产环境建议配合监控观察文件数与查询 P95 延迟，再回头微调。

七、与 Iceberg / Hudi 怎么选

三者在「湖仓」赛道各有侧重，选型时看你的主链路是流还是批：

维度	Apache Paimon	Apache Iceberg	Apache Hudi	传统 Hive
流批一体	原生流批双模	偏批，近实时写	近实时（MOR）	仅批
主键更新	原生 LSM 更新	需 rewrite / merge	原生 upsert	需全量覆盖
实时入湖延迟	秒级	分钟级	分钟级	小时级（T+1）
典型适用	实时数仓、CDC 入湖	海量分析、归档	近实时更新	离线数仓

经验法则：上游是 CDC/Kafka、要实时最新值且下游还要跑批，优先 Paimon；以对象存储上的超大数据集离线分析为主、强调开放生态，选 Iceberg；已经重度使用 Spark/HDFS 且要 upsert，Hudi 也很成熟。

八、生产落地检查清单

检查项	建议
存储底座	用对象存储（S3/OSS）或 HDFS 做 warehouse，别用本地盘，否则 TaskManager 漂移丢数据
Bucket 数	按数据量与并发设 4~32，过小会热点、过大则小文件过多
Changelog	有流式下游时设 changelog-producer=input 或 full，否则下游读不到变更
小文件合并	开启 compaction（默认后台自动），高写入场景调大 write-buffer 与合并线程
新鲜度	流式读默认追最新；对账场景用 batch 模式保证全量一致
CI 与回滚	DDL 纳入 GitHub Actions 版本化，配合 snapshot 时间旅行可随时回看

Apache Paimon 把「流」和「湖」之间的裂缝填平了：一张表既是实时管道，也是分析底座。当你再被「实时 GMV 和昨天对不上」这类问题追问时，答案就在这张既能写、也能查的 Paimon 表里。环境搭建可复用本站 Docker Compose 实战 的底座，先把 Flink 与 Kafka 跑顺，再让数据自然地「流」进湖中。