Flink 2.3.0 从理论到实践 —— 第 15 章 Paimon & Doris 湖仓写入

Flink 2.3.0 从理论到实践 —— 第 15 章 Paimon & Doris 湖仓写入

课程定位：第五篇数据集成收官。报文经 Kafka 入湖、维表经 CDC 入湖后，本章讲事实表如何写 Paimon、结果如何写 Doris。讲解 Paimon 主键表/Append 表/Changelog Producer、批读/流读/Lookup 三种读取、Catalog 注册、Doris Stream Load/2PC/部分列更新、JDBC vs Doris Connector 读取选型、端到端 Exactly-Once 链路、Flink+Paimon+Doris 湖仓协同——这是"湖上做计算、Doris 做服务"架构落地的核心章节。

版本基线：Flink 2.3.0 + Paimon 1.4.x（paimon-flink-2.3）+ Doris 4.1

配套教程：Apache Paimon 1.4 从理论到实践、Apache Doris 4.1 从理论到实践（基础理论）

章节导读

• 15.1 Paimon 写入：主键表 / Append 表 / Changelog Producer
• 15.2 Paimon 读取：批读 / 流读 / Lookup
• 15.3 Paimon Catalog 注册
• 15.4 Doris Sink：Stream Load / 2PC / 部分列更新
• 15.5 Doris Read：JDBC vs Doris Connector
• 15.6 端到端 Exactly-Once 链路
• 15.7 湖仓一体：Flink + Paimon + Doris 协同
• 15.8 本章小结与下章预告

15.1 Paimon 写入：主键表 / Append 表 / Changelog Producer

15.1.1 两类表选型

┌──────────────────────────────────────────────────────────────┐
│ Paimon 两类表 │
└──────────────────────────────────────────────────────────────┘

① Primary-Key Table(主键表) ② Append-Only Table(追加表)
有主键,支持 upsert/删除 无主键,只追加
事实表/维表/结果表 日志/审计/只写不改
LSM 结构 + changelog 文件追加

项目选型：车联网 ODS 企标报文用主键表——主键 (vin, collect_time, dt) 报文级幂等，重放不翻倍。纯审计日志（只追加不改）可用 Append 表省主键开销。

15.1.2 主键表建表（项目实战）

CREATE CATALOG paimon WITH (
'type' = 'paimon',
'metastore' = 'hive',
'uri' = 'thrift://hm-1:9083,thrift://hm-2:9083',
'warehouse' = 'hdfs:///warehouse/tablespace/managed/hive'
);

CREATE TABLE IF NOT EXISTS paimon.tsp_ods.ods_vehicle_qb_msg_report_original (
vin STRING,
collect_time BIGINT,
json_data STRING,
dt STRING,
PRIMARY KEY (vin, collect_time, dt) NOT ENFORCED
) PARTITIONED BY (dt) WITH (
'bucket' = '6',
'bucket-key' = 'vin',
'file.format' = 'parquet',
'file.compression' = 'zstd',
'file.compression.zstd-level' = '3',
'write-buffer-size' = '256 mb',
'target-file-size' = '128 mb',
'changelog-producer' = 'full-compaction',
'compaction.max.file-num' = '5',
'dynamic-partition.create-history-partition' = 'true'
— 并配 history_partition_num(允许创建的历史分区数量,按回补窗口设置)
);

设计决策（项目实测）：

15.1.3 Changelog Producer

主键表要让下游流读到 UPDATE/DELETE 变更，需配 changelog-producer：

项目硬约束：用 changelog-producer=full-compaction + Checkpoint 30s。下游流读可见性 ≤ 30s（compaction 随 checkpoint 触发）。不用 lookup/input——它们对每条更新查历史/要求输入即 changelog，成本高；车联网实时容忍 30s 延迟，full-compaction 性价比最高。

15.1.4 Merge Engine（合并引擎）

— 宽表拼接:多个作业各写一部分列
CREATE TABLE dwd_vehicle_wide (...) WITH (
'merge-engine' = 'partial-update',
'changelog-producer' = 'lookup',
'fields.speed.aggregate-function' = 'last_non_null_value'
);

用途：DWD 宽表由多个作业拼接（报文解析作业写基础列、电池作业写电池列），partial-update 列级合并互不覆盖。

15.1.5 Append 表

CREATE TABLE ods_audit_log (
vin STRING,
op_time TIMESTAMP(3),
action STRING,
dt STRING
) PARTITIONED BY (dt) WITH (
'bucket' = '6',
'file.format' = 'parquet',
'file.compression' = 'zstd'
— 无 PRIMARY KEY → Append 表
);

Append 表写入不依赖 Checkpoint（追加即可见），但无法 upsert/去重。只写不改的日志场景用。

15.1.6 动态分区历史分区（回补防丢数据）

项目硬约束：回补历史分区时报 no partition for this tuple 且 Sink 静默丢数据（比报错更危险）。根因是动态分区表默认只自动建"当前及未来"分区。必须配：

'dynamic-partition.create-history-partition' = 'true'
— 并配 history_partition_num(允许创建的历史分区数量)

这样回补历史数据时自动建历史分区，不再静默丢数据。

15.2 Paimon 读取：批读 / 流读 / Lookup

15.2.1 三种读取模式

┌──────────────────────────────────────────────────────────────┐
│ Paimon 三种读取模式 │
└──────────────────────────────────────────────────────────────┘

① Batch Read(批读) ② Streaming Read(流读) ③ Lookup(点查)
扫快照,读完即退 持续读 changelog 按 key 查最新值
离线作业 实时作业 维表 JOIN

15.2.2 流读：scan.mode 选型（铁律）

— ★ 实时作业:只读最新快照起
SELECT * FROM paimon.tsp_dws.dws_vehicle_trip_segment_rt
/*+ OPTIONS('scan.mode' = 'latest') */;

项目硬约束：实时作业源表必须 scan.mode='latest'。默认 latest-full 会全量回放表历史——ODS 起量后直接把实时作业 OOM（冷启动"不可用"）。历史数据是离线的事，实时只从当前快照开始（流批分工）。

15.2.3 批读：离线跑批

— 批模式离线作业,只扫目标分区
SET 'execution.runtime-mode' = 'batch';

INSERT OVERWRITE paimon.tsp_dws.dws_vehicle_trip_daily PARTITION (dt = '__DT__')
SELECT vin, COUNT(*), SUM(distance)
FROM paimon.tsp_ods.ods_vehicle_qb_msg_report_original
WHERE dt = '__DT__'
GROUP BY vin;

分区裁剪：WHERE dt='__DT__' 谓词下推，批读只扫目标分区；Paimon 批读还会做 Bucket 合并，每日跑批只处理 T 日增量。

15.2.4 Lookup Join：维表点查

— 主流流读 + Paimon 维表 Lookup
SELECT m.vin, m.collect_time, d.car_series, d.car_type
FROM paimon.tsp_ods.ods_vehicle_qb_msg_report_original_rt AS m
LEFT JOIN paimon.tsp_ods.ods_iov_veh_t_cn_car_info
FOR SYSTEM_TIME AS OF m.proctime() AS d
ON m.vin = d.vin;

项目硬约束：维表（车型/车系/组织）经 Flink CDC 入 Paimon 后，实时作业 Lookup、离线批作业批读，一份维表两链路共享。Lookup 缓存 TTL 设 1h（维表变更不频繁）。

15.2.5 时间戳单位坑

项目硬约束：Paimon 表的 COLLECT_TIME 若是 13 位毫秒时间戳，用 TO_TIMESTAMP_LTZ 转换时需先除以 1000（或乘 1000 前先判断位数）。车联网报文是 10 位秒级，乘 1000 转毫秒；若上游混入 13 位毫秒，直接乘 1000 会得到 23 位溢出时间。建议建表统一时间戳单位，或在计算列做位数判断：

event_time AS TO_TIMESTAMP_LTZ(
CASE WHEN collect_time > 10000000000 THEN collect_time ELSE collect_time * 1000 END, 3)

15.3 Paimon Catalog 注册

15.3.1 三种 Metastore

15.3.2 生产注册（Hive Metastore）

CREATE CATALOG paimon WITH (
'type' = 'paimon',
'metastore' = 'hive',
'uri' = 'thrift://hm-1:9083,thrift://hm-2:9083',
'warehouse' = 'hdfs:///warehouse/tablespace/managed/hive'
);

项目硬约束：离线批 / 实时流共用同一个 Hive Metastore Catalog——这是"同引擎同存储，不存在跨链路同步"的元数据基础。离线 INSERT OVERWRITE 与实时 INSERT INTO 写同一张表的同一份元数据，没有异构存储搬运。Doris 也通过同一个 HMS 用 Paimon Catalog 联邦直查。

15.3.3 jar 加载铁律

项目硬约束：paimon-flink-2.3 jar 只放集群 ${FLINK_HOME}/lib 一份，提交脚本绝不传 -j。-j 与 lib 双份加载同一连接器类（双 classloader），建 Paimon Catalog 报：

ServiceConfigurationError: … not a subtype

判据：连接器 jar 恰好 1 个 + Paimon 版本唯一。

15.4 Doris Sink：Stream Load / 2PC / 部分列更新

15.4.1 Doris 表模型前置

项目硬约束：Doris 结果表必须用 Unique Key 模型 + Merge-on-Write（enable_unique_key_merge_on_write=true）。DUPLICATE KEY 模型不做去重，Flink 重放会产生重复行；Merge-on-Write 保证主键实时合并，点查性能好。

15.4.2 Doris 建表（告警表实战）

CREATE TABLE tsp_alert.dwd_vehicle_fence_alert (
vin VARCHAR(17) NOT NULL COMMENT '车架号',
session_id VARCHAR(64) NOT NULL COMMENT '会话ID(进出配对)',
event_type VARCHAR(16) NOT NULL COMMENT 'ENTER/EXIT',
fence_id VARCHAR(64) NOT NULL COMMENT '围栏ID',
alert_time DATETIME NOT NULL COMMENT '告警时间(秒级)',
message VARCHAR(512) NULL COMMENT '告警内容',
dt VARCHAR(10) NOT NULL DEFAULT '',
— 倒排索引(项目硬约束:vin/event_type/message)
INDEX idx_vin (vin) USING INVERTED COMMENT '车架号倒排索引',
INDEX idx_event_type (event_type) USING INVERTED COMMENT '事件类型索引',
INDEX idx_message (message) USING INVERTED PROPERTIES('parser'='chinese') COMMENT '告警内容全文索引'
)
UNIQUE KEY (vin, session_id, event_type)
DISTRIBUTED BY HASH(vin) BUCKETS 6
PROPERTIES (
'enable_unique_key_merge_on_write' = 'true',
'function_column.sequence_col' = 'alert_time',
'compaction_policy' = 'time_series',
'replication_num' = '3'
);

项目硬约束①：围栏告警必须同时输出 enter 和 exit 两类告警；ActiveSession 用 sessionId 做进出事件 upsert 配对（不是 vin 单主键，否则 enter 被 exit 覆盖）。

项目硬约束②：Doris 必须为 vin、event_type、message 建倒排索引——高频点查 + 告警内容全文检索。

项目硬约束③：围栏表 vin 非空时只对指定车辆生效。

15.4.3 Flink 写 Doris（DDL）

CREATE TABLE doris_sink_fence_alert (
vin STRING,
session_id STRING,
event_type STRING,
fence_id STRING,
alert_time TIMESTAMP(3),
message STRING,
dt STRING,
PRIMARY KEY (vin, session_id, event_type) NOT ENFORCED
) WITH (
'connector' = 'doris',
'fenodes' = 'doris-fe-1:28030,doris-fe-2:28030,doris-fe-3:28030',
'table.identifier' = 'tsp_alert.dwd_vehicle_fence_alert',
'username' = '${doris.user}',
'password' = '${doris.pwd}',
'sink.label-prefix' = 'fence-alert-sink',
'sink.properties.format' = 'json',
'sink.properties.read_json_by_line' = 'true',
'sink.enable-2pc' = 'true', — ★ 两阶段提交
'sink.properties.partial_columns' = 'true', — ★ 部分列更新
'sink.properties.max_filter_rows' = '100',
'sink.buffer-flush.max-rows' = '500',
'sink.buffer-flush.interval' = '5s'
);

15.4.4 Stream Load 与 2PC

┌──────────────────────────────────────────────────────────────┐
│ Doris Stream Load 两阶段提交 │
└──────────────────────────────────────────────────────────────┘

① Flink 攒批(500 行/5s)
│
▼
② 发起事务,PUT 数据(label=prefix-checkpointId)
│
▼
③ Checkpoint 成功 ──► commit 事务(数据可见)
Checkpoint 失败 ──► abort 事务(回滚)

Label 幂等：Stream Load 的 label 唯一，同一 label 重复提交只生效一次——故障重试不产生重复。label 由 label-prefix + checkpointId 组成，天然幂等。

15.4.5 部分列更新铁律

项目硬约束①：partial_columns=true 时只写 DDL 声明的列。Doris 表所有 NOT NULL 列必须有默认值或允许 NULL——否则部分列写入缺值直接失败。建表时 dt 这类列给 DEFAULT ''。

项目硬约束②：unique_key_update_mode 不能用 UPDATE_PARTIAL_COLUMNS（无效值）。合法值仅 UPSERT / UPDATE_FIXED_COLUMNS / UPDATE_FLEXIBLE_COLUMNS。用 UPDATE_FLEXIBLE_COLUMNS 模式时同样要求 NOT NULL 列有默认值或可空。

项目硬约束③（DATETIME 格式）：Doris DATETIME 只收 yyyy-MM-dd HH:mm:ss（秒级，无毫秒/纳秒）。Flink TIMESTAMP(3) 序列化必须截断到秒：

DATE_FORMAT(alert_time, 'yyyy-MM-dd HH:mm:ss') AS alert_time

否则 Stream Load 报时间格式错误。

项目硬约束④（sinkTo API）：DataStream API 写 Doris 必须用 stream.sinkTo(dorisSink)，不能用 addSink(SinkFunction)——Flink 1.20+ / 2.x 已移除 SinkFunction API。SQL Connector 不受影响。

15.4.6 Colocate Join

项目硬约束：高频关联的 Doris 表配 Colocate Join——相同分桶列 + 相同桶数 + 同一 Colocate Group，JOIN 时本地完成免网络 shuffle。车联网结果表统一 DISTRIBUTED BY HASH(vin) BUCKETS 6 + 同一 group。

15.5 Doris Read：JDBC vs Doris Connector

15.5.1 选型决策（血泪坑）

Flink 读 Doris
│
┌──────────────┴──────────────┐
▼ ▼
connector='jdbc' connector='doris'
(MySQL 协议 29030) (Arrow 协议)
│ │
✅ 读(★ 项目采用) ❌ 读(DATE 类型坑)
✅ 写(Stream Load)

项目硬约束（Doris 4.1 Arrow 坑）：Doris 4.1 的 Arrow 协议把 DATE 类型返回为带时区的 TIMESTAMPSECTZ，Doris Connector 只认 DATEDAY/VARCHAR，读表直接报：

FLINK type is DATEV2, but arrow type is TIMESTAMPSECTZ

铁律：读 Doris 用 connector='jdbc'，写 Doris 才用 connector='doris'。JDBC 走 MySQL 协议，规避 Arrow 类型兼容问题。

15.5.2 JDBC 读 Doris 配置

CREATE TABLE jdbc_doris_result (
vin STRING, session_id STRING, event_type STRING,
alert_time TIMESTAMP(3)
) WITH (
'connector' = 'jdbc',
— ★ 两处时区配置缺一不可
'url' = 'jdbc:mysql://doris-fe:29030/tsp_alert?serverTimezone=Asia/Shanghai',
'table-name' = 'dwd_vehicle_fence_alert',
'username' = '${doris.user}',
'password' = '${doris.pwd}',
'lookup.cache.max-rows' = '10000',
'lookup.cache.ttl' = '1 h'
);

项目硬约束（时区静默写错）：JDBC 读 Doris 的 TIMESTAMP 若时区缺失，整体偏 8 小时且静默写错（不报错）。两处配置缺一不可：

15.5.3 两种读方式的用途

15.6 端到端 Exactly-Once 链路

15.6.1 完整链路 EOS 组成

┌──────────────────────────────────────────────────────────────┐
│ Kafka → Flink → Paimon/Doris 端到端 EOS │
└──────────────────────────────────────────────────────────────┘

Kafka Source Flink Checkpoint Sink
┌──────────┐ ┌──────────────┐ ┌────────────┐
│ 位点提交 │◄────────►│ barrier 对齐 │◄──────►│ Paimon: │
│ (group- │ │ (Chandy-Lamport)│ │ checkpoint │
│ offsets) │ │ │ │ commit │
└──────────┘ └──────────────┘ ├────────────┤
│ Doris: │
│ 2PC commit │
└────────────┘
全链路同一 Checkpoint 周期:30s

15.6.2 幂等兜底

即便 EOS 生效，Checkpoint 间隔内故障仍会重放最近一批数据，靠各层幂等兜底：

项目实践：Paimon 主键 (vin, collect_time, dt) 报文级幂等；Doris Unique Key + sessionId 配对。Checkpoint 30s + 两层幂等，故障重放不产生重复行。

15.6.3 Checkpoint 配置（对齐可见性）

SET 'execution.checkpointing.interval' = '30 s'; — Paimon/Doris 提交周期
SET 'execution.checkpointing.mode' = 'EXACTLY_ONCE';
SET 'execution.checkpointing.timeout' = '10 min';
SET 'execution.checkpointing.min-pause' = '5 s';
SET 'execution.checkpointing.max-concurrent-checkpoints' = '1';

项目硬约束：Checkpoint 30s 是全链路对齐的关键——Paimon full-compaction changelog 可见性 ≤ 30s，Doris 2PC 提交周期 30s。过短→反压；过长→故障重放数据多、实时可见性差。

15.7 湖仓一体：Flink + Paimon + Doris 协同

15.7.1 三层分工

┌──────────────────────────────────────────────────────────────┐
│ Flink + Paimon + Doris 湖仓分工 │
└──────────────────────────────────────────────────────────────┘

Flink(计算) Paimon(湖底座) Doris(服务加速)
┌─────────┐ ┌───────────────┐ ┌──────────────┐
│ 实时流作业│ ──────► │ ODS/DWD/DWS │ ◄─联邦── │ Paimon │
│ 离线批作业│ ──────► │ 流批一套存储 │ Catalog │ 直查(当天) │
└─────────┘ └───────────────┘ ├──────────────┤
│ MV(历史加速) │
Flink ──2PC写──► │ 分流 VIEW │
└──────────────┘

15.7.2 为什么服务层还要 Doris

Paimon 是 scan 式查询，扛不住 BI 的"筛选+聚合+高并发点查"交互延迟。Doris 承担加速：

• 历史数据：在离线 Paimon 表之上建异步物化视图（BUILD DEFERRED，每日凌晨分区级 REFRESH），查询走 MV 本地副本毫秒级；
• 当天数据：Doris 通过 Paimon Catalog 联邦直查湖里 _rt 表最新快照；
• 统一入口：VIEW v_* 按 dt 分流，BI 只换表名，SQL 零改动。

15.7.3 Doris Paimon Catalog 联邦

— Doris 内注册 Paimon Catalog,直查湖表
CREATE CATALOG paimon_lake PROPERTIES (
'type' = 'paimon',
'paimon.catalog.type' = 'hive',
'hive.metastore.uris' = 'thrift://hm-1:9083,thrift://hm-2:9083',
'warehouse' = 'hdfs:///warehouse/tablespace/managed/hive'
);

— Doris 内直查湖里实时表(当天)
SELECT * FROM paimon_lake.tsp_dws.dws_vehicle_trip_daily_rt
WHERE dt >= DATE_FORMAT(CURRENT_DATE(), '%Y-%m-%d');

15.7.4 分流 VIEW（R3 契约）

CREATE VIEW tsp_dws.v_dws_vehicle_trip_daily AS
SELECT * FROM mv_dws_vehicle_trip_daily — 历史走 MV
WHERE dt < DATE_FORMAT(CURRENT_DATE(), '%Y-%m-%d')
UNION ALL
SELECT * FROM paimon_lake.tsp_dws.dws_vehicle_trip_daily_rt — 当天走联邦直查
WHERE dt >= DATE_FORMAT(CURRENT_DATE(), '%Y-%m-%d');

关键细节：UNION 分流只能放 VIEW，不能放 MV——VIEW 定义在查询时求值，MV 内 SQL 在建表时固化。切流=BI 换表名（30 秒）；回滚=删 VIEW 实时分支，BI 无感；实时链路整体停掉不影响对外查询（历史走离线）——分名方案最大的安全优势。

15.7.5 物理分名与分区契约

核心结论（项目经验）：

15.7.6 数据治理

反面教训：hist/ads_stat 这类"当前值快照表"（PK=vin、整表覆盖、vin 点查）加 dt 分区是负优化——主键被迫变 (vin,dt) 语义漂移、行数×保留天数、MV 全量刷新成本×天数。判定线：能配 TTL 过期的表才适合按天分区。

15.8 本章小结与下章预告

本章小结

┌────────────────────────────────────────────────────────────────┐
│ 第 15 章 要点回顾 │
└────────────────────────────────────────────────────────────────┘

✓ Paimon 写入:
主键表(upsert/去重) vs Append 表(只追加)
项目: PK(vin,collect_time,dt) / bucket 6 by vin / zstd-3
changelog-producer=full-compaction(可见性 ≤ 30s)
merge-engine: deduplicate/partial-update/aggregation
★ 回补开 dynamic-partition.create-history-partition=true

✓ Paimon 读取:
批读(BATCH,分区裁剪)/流读(STREAMING)/Lookup(维表)
★ 实时源 scan.mode=latest(非 latest-full,防 OOM)
维表一份:实时 Lookup + 离线批读共享
★ 13 位毫秒时间戳要除 1000

✓ Catalog:
metastore=hive,实时/离线/Doris 三方共享一个 HMS
★ jar 只放 lib,绝不传 -j(防 not a subtype)

✓ Doris Sink:
★ Unique Key + Merge-on-Write
★ 围栏 sessionId 配对 enter/exit,倒排索引 vin/event_type/message
Stream Load + 2PC + label 幂等
★ 部分列更新:NOT NULL 列要有默认值/可空
★ update mode 只能 UPSERT/UPDATE_FIXED/FLEXIBLE_COLUMNS
★ DATETIME 秒级,DataStream 用 sinkTo
Colocate Join:同列同桶同 group

✓ Doris Read:
★ 读用 connector='jdbc'(Arrow DATE 坑)
★ 写用 connector='doris'
★ URL serverTimezone + SET table.local-time-zone(防偏 8h)

✓ 端到端 EOS:
Kafka 位点 + Checkpoint 30s + Paimon commit + Doris 2PC
幂等兜底:Paimon PK / Doris Unique Key+label

✓ 湖仓协同:
Flink 计算 / Paimon 湖底座 / Doris 服务加速
Doris Paimon Catalog 联邦直查当天, MV 加速历史
★ 分流只能放 VIEW 不能放 MV
分名 _rt + R1/R2/R3 + 种子校准
★ 同步工具是伪需求,种子校准才是真需求

下章预告

第 16 章 生产部署与监控：湖仓链路搭完，如何上生产。讲解生产部署模式选型、JM/TM/Slot/内存资源规划、Checkpoint 与 Savepoint 管理、Prometheus + Grafana 监控体系、关键指标（Checkpoint 时长/背压/状态大小/延迟）、告警规则与 SOP、日志管理与排查——这是第六篇部署运维开篇。

官方参考资料

• Paimon Flink Connector：https://paimon.apache.org/docs/master/flink/
• Paimon Catalog：https://paimon.apache.org/docs/master/maintenance/catalog/
• Paimon Primary Key Table：https://paimon.apache.org/docs/master/concepts/primary-key-table/
• Doris Flink Connector：https://doris.apache.org/docs/ecosystem/flink-doris-connector/
• Doris Stream Load：https://doris.apache.org/docs/data-operate/import/stream-load-manual/
• Doris Paimon Catalog：https://doris.apache.org/docs/lakehouse/multi-catalog/paimon/
• Doris Unique Key：https://doris.apache.org/docs/table-design/data-model/unique/