零脏读的 Serverless 湖仓:我们如何用 DuckDB 解决一致性难题
Forma 工程博客 · 系列第三篇(完结)
TL;DR
"Lakehouse"听起来很美好,但大家心里都有同一个疑问:我怎么知道查出来的数据不是脏的?
这篇文章解释 Forma 如何用 Anti-Join(反连接)加 Dirty Set(脏数据集)机制,让联邦查询读不到未提交或不一致的数据。PostgreSQL 负责当下,DuckDB + Parquet 负责历史,两者协同,零脏读。
为什么需要 Lakehouse?
前两篇文章解决了 OLTP 场景的问题。第一篇用热表加 JSON Schema 实现了 AI-Ready 的灵活存储,第二篇用 CTE + JSON_AGG 消灭了 N+1 查询,延迟从 1 秒降到 25 毫秒。
但有一个问题一直没正面回答:当数据量达到亿级,PostgreSQL 单机扛不住怎么办?
即使有了热表索引,当 EAV 表膨胀到 1 亿行、Parquet 化的历史数据达到 TB 级,单机 PostgreSQL 的内存和 I/O 都会成为瓶颈。
历史数据的访问模式也和实时数据完全不同:
| 数据类型 | 访问频率 | 访问模式 | 典型场景 | 占查询量 |
|---|---|---|---|---|
| 最近 7 天 | 每秒数百次 | 点查、过滤、分页 | 日常操作 | ~80% |
| 7-90 天 | 每天数十次 | 批量导出、报表 | 月度分析 | ~15% |
| 90 天以上 | 每月数次 | 全量扫描、聚合 | 年度审计 | ~5% |
为最近 7 天的数据优化,却让它和 3 年的历史数据挤在同一张表里,两头都浪费资源。
湖仓架构的诱惑
冷热分离是明显的解法。热数据留在 PostgreSQL,享受事务一致性和低延迟索引;冷数据导出到 Parquet 文件存在 S3,用 OLAP 引擎查询。
架构图看起来很清爽:
┌─────────────────────────────────────────────────────────────┐
│ Query Router │
└─────────────────────────────────────────────────────────────┘
│ │
▼ ▼
┌───────────────────┐ ┌───────────────────┐
│ PostgreSQL │ │ DuckDB │
│ (Hot Data) │ │ (Cold Data) │
│ Last 7 days │ │ Parquet on S3 │
└───────────────────┘ └───────────────────┘然后每个听到"Lakehouse"的工程师心里都会冒出同一个声音:
"等等,如果同一条记录同时存在于 PostgreSQL 和 Parquet 里,我查出来的到底是哪个版本?如果 PostgreSQL 里的数据还没同步到 Parquet,我会不会读到旧数据?或者更糟,重复数据?"
这就是一致性恐惧,也是阻止很多团队采用 Lakehouse 架构的最大心理障碍。
"等等,EAV 不是反模式吗?"
从历史上看确实是。EAV(Entity-Attribute-Value,实体-属性-值)被认为是反模式,理由也充分。传统的 EAV 实现查询性能糟糕,N+1 问题把简单查询变成数千次数据库往返;没有类型安全,一切都变成字符串,整数比较和日期排序都没了;代码难以维护,动态 pivot 查询散落各处;也没有可行的索引策略,因为你无法索引"任意可能的属性",每次查询都得全表扫描。
如果你曾被 EAV 坑过,这份怀疑是有道理的。
Forma 如何驯服它
Forma 用具体的技术选择逐个解决这些问题:
| EAV 问题 | Forma 的解决方案 | 在哪里介绍 |
|---|---|---|
| N+1 查询噩梦 | CTE + JSON_AGG(单次往返) | 第二篇 |
| 没有类型安全 | 写入时 JSON Schema 校验 | 第一篇 |
| 没有索引 | 热表 + B-tree 索引用于高频字段 | 第一篇 |
| 湖仓中的脏数据 | 脏数据集隔离区(Anti-Join) | 本文(下方) |
数据隔离区
脏数据集(Dirty Set)就是数据的隔离区。把它想象成机场安检:数据被放行(flushed_at > 0)之前,它待在候机区(PostgreSQL);一旦放行,就可以前往目的地(Parquet)。让系统可信的正是这种显式的状态追踪,而不是时间戳或启发式规则。
任何被修改但还没同步到冷存储的记录都处于隔离状态。查询会从 PostgreSQL 这个真相来源取它,而不是从可能过时的 Parquet。没有猜测,没有竞态条件,也不用指望时间戳恰好是对的。
串起来看
EAV 很强大,但也容易用错,所以我们在它周围建了四道护栏:通过 JSON Schema 做类型校验(第一篇)、为那 20% 占据 80% 查询量的属性建索引热字段(第一篇)、通过 CTE + JSON_AGG 做单查询聚合(第二篇),以及通过脏数据集做显式同步状态追踪(本文)。
结果是灵活但不混乱。AI 应用可以自由演进数据结构,不用背负 EAV 历史坏名声的包袱。
一致性恐惧的根源
拿一条具体记录 row_id = 123 来看:
- 09:00:用户创建这条记录,写入 PostgreSQL
- 09:05:CDC 作业将它导出到 Parquet
- 09:10:用户更新它,PostgreSQL 中的值变了
- 09:15:用户发起查询
09:15 的查询应该返回哪个版本?
| 数据源 | row_id | 版本 | 状态 |
|---|---|---|---|
| PostgreSQL | 123 | v2 | 最新(09:10 更新) |
| Parquet | 123 | v1 | 过时(09:05 导出) |
如果查询引擎天真地合并两个数据源,用户会拿到重复(同一条记录出现两次,v1 和 v2)、脏读(返回过时的 v1),或者幻读(有时 v1,有时 v2,取决于查询时机)。这些都不可接受。
为什么简单的时间戳比较不够?
最直接的想法是用 updated_at 时间戳做去重:
SELECT * FROM (
SELECT *, 'pg' AS source FROM postgres_data
UNION ALL
SELECT *, 's3' AS source FROM parquet_data
)
WHERE row_number() OVER (PARTITION BY row_id ORDER BY updated_at DESC) = 1这个方案有几个致命问题。PostgreSQL 和 CDC 作业的时钟可能有毫秒级偏差。记录在导出中途被更新时,两边的 updated_at 可能相同。而如果记录在 PostgreSQL 中被删除,Parquet 里的旧版本会"复活"。
时间戳比较是乐观的,它假设时间戳能完美反映数据新旧。在分布式系统中,这种假设很危险。
Forma 的解法:Anti-Join + Dirty Set
Forma 采用悲观策略:不信任时间戳,只信任状态。
核心思想是引入一个脏数据集(Dirty Set):
如果一条记录在 PostgreSQL 中还没落盘到 Parquet,那么无论 Parquet 里有没有这条记录,都忽略 Parquet 版本,只用 PostgreSQL 版本。
change_log 表:脏数据的源头
Forma 在 PostgreSQL 中维护一张 change_log 表:
CREATE TABLE change_log (
id BIGSERIAL PRIMARY KEY,
schema_id UUID,
row_id UUID,
op SMALLINT, -- 1=INSERT, 2=UPDATE, 3=DELETE
created_at BIGINT, -- 变更时间戳
flushed_at BIGINT -- 导出时间戳;0 = 未导出
);关键字段是 flushed_at。等于 0,表示这条变更还没同步到 Parquet,数据是脏的;大于 0,表示已经同步,数据是干净的。
查询时的 Anti-Join 逻辑
当用户发起查询时,Forma 的 DuckDB 查询引擎执行以下逻辑:
SQL 实现:
-- 步骤 1: 获取脏数据集(还没落盘的 row_id)
dirty_ids AS (
SELECT row_id
FROM change_log
WHERE flushed_at = 0 AND schema_id = $SCHEMA_ID
),
-- 步骤 2: 从 Parquet 读取数据,但排除脏数据集中的记录
s3_clean AS (
SELECT *
FROM read_parquet('s3://bucket/data/*.parquet')
WHERE row_id NOT IN (SELECT row_id FROM dirty_ids) -- Anti-Join!
),
-- 步骤 3: 从 PostgreSQL 读取脏数据(最新版本)
pg_hot AS (
SELECT *
FROM postgres_scan('SELECT * FROM entity_main WHERE ...')
WHERE row_id IN (SELECT row_id FROM dirty_ids)
),
-- 步骤 4: 合并
SELECT * FROM s3_clean
UNION ALL
SELECT * FROM pg_hot用公式表示:
$$Result = (Parquet_{data} \setminus DirtySet) \cup PostgreSQL_{hot}$$
翻译成人话:Parquet 数据只保留那些已经落盘、没有更新版本的记录;PostgreSQL 数据只保留那些还没落盘、或者刚刚更新的记录;两者取并集,保证每条记录只出现一次,且是最新版本。
为什么这个方案是"悲观"且安全的?
新旧由"是否已同步"这个明确状态决定,而不是由时间戳决定:
| 场景 | Dirty Set | Parquet | PostgreSQL | 返回 |
|---|---|---|---|---|
| 记录只在 PG | row_id ∈ Dirty | 无 | 有 | PG 版本 |
| 记录已同步,无更新 | row_id ∉ Dirty | 有 | 有(相同) | Parquet 版本 |
| 记录已同步,有更新 | row_id ∈ Dirty | 有(旧) | 有(新) | PG 版本 |
| 记录在 PG 删除 | row_id ∈ Dirty | 有(旧) | 无 | 不返回 |
无论哪种场景,用户看到的都是最新、一致的数据。
类比:快递还在路上的订单
想象你经营一家网店,有两本账本。本地账本(PostgreSQL)实时记录每笔订单;云端账本(Parquet)每天晚上从本地账本同步一次。
现在有人问:"今天的销售额是多少?"
把两本账本的数字直接加起来是错的,因为还在同步中的订单会被重复计算。
正确的做法是:先看哪些订单还没同步到云端,这就是脏数据集;云端账本的数据,排除这些还在路上的订单;本地账本的数据,只算这些还在路上的订单;两者相加。
这就是 Anti-Join 加 Dirty Set 的逻辑。
CDC 流程:数据如何从 PostgreSQL 流向 Parquet
写入时:记录变更
每次写入 entity_main 或 eav_data 时,同步插入 change_log:
-- 应用写入数据
INSERT INTO entity_main (...) VALUES (...);
INSERT INTO eav_data (...) VALUES (...);
-- 记录变更(flushed_at = 0 表示未导出)
INSERT INTO change_log (schema_id, row_id, op, created_at, flushed_at)
VALUES ($schema_id, $row_id, 1, now(), 0);CDC 作业:增量导出
CDC 作业定期运行,默认每分钟一次:
-- 1. 找出待导出的 row_id
SELECT DISTINCT row_id FROM change_log
WHERE schema_id = $SCHEMA_ID AND flushed_at = 0;
-- 2. 读取完整记录,展平 EAV 为宽表
SELECT m.row_id, m.text_01 AS name, m.integer_01 AS age, ...
FROM entity_main m
LEFT JOIN eav_data e ON m.row_id = e.row_id
WHERE m.row_id IN ($PENDING_IDS);
-- 3. 写入 Parquet
COPY (...) TO 's3://bucket/delta/<uuid>.parquet';
-- 4. 标记已导出
UPDATE change_log SET flushed_at = now()
WHERE row_id IN ($PENDING_IDS) AND flushed_at = 0;数据流全景图
┌─────────────────────────────────────────────────────────────────────┐
│ Query Path │
│ DuckDB: (Parquet - DirtySet) ∪ (PostgreSQL ∩ DirtySet) │
└─────────────────────────────────────────────────────────────────────┘故障模式与自愈
分布式系统总会出故障。Forma 要做的是在故障发生时不让数据损坏,下面逐个场景来看。
CDC 崩溃恢复
任何同步系统中最危险的时刻是导出中途崩溃。如果 CDC 任务在写入 S3 后、更新 flushed_at 之前挂掉会怎样?
救命的是写入顺序:
┌─────────────────────────────────────────────────────────────────────┐
│ CDC 任务执行顺序 │
│ │
│ 1. BEGIN TRANSACTION (PostgreSQL) │
│ 2. SELECT * FROM entity_main WHERE row_id IN (dirty_ids) │
│ 3. Write to S3 Parquet ◄─── 如果在这里崩溃,Parquet 有数据 │
│ 4. UPDATE change_log SET flushed_at = now() ◄─── 但 PG 没有更新 │
│ 5. COMMIT │
└─────────────────────────────────────────────────────────────────────┘假设在步骤 3 之后、步骤 4 之前崩溃:
| 组件 | 崩溃后状态 |
|---|---|
| S3 Parquet | 包含导出的数据 |
| change_log.flushed_at | 仍然是 0(未更新) |
| 下次查询 | 将从 PostgreSQL 获取(正确!) |
这是设计上的安全。因为 flushed_at 只在 S3 写入成功之后才更新,崩溃会让记录保留在 Dirty Set 中。查询引擎看到 flushed_at = 0,判定记录是脏的,从 PostgreSQL 获取,忽略那个孤立的 Parquet 文件,它会在下次成功导出时被覆盖。
┌─────────┐ ┌─────────┐ ┌─────┐ ┌─────────┐
│ CDC 任务│ │ PG │ │ S3 │ │ 查询 │
└────┬────┘ └────┬────┘ └──┬──┘ └────┬────┘
│ 1. 读取脏 ID │ │ │
│◄───────────────────│ │ │
│ │ │ │
│ 2. 读取完整数据 │ │ │
│◄───────────────────│ │ │
│ │ │ │
│ 3. 写入 Parquet │ │ │
│──────────────────────────────────────▶│ │
│ │ │ │
│ ╔═══════════════╧══════════════════╧════╗ │
│ ║ ⚡ 在这里崩溃 ║ │
│ ╚═══════════════╤══════════════════╤════╝ │
│ │ │ │
│ │ │ 4. 查询到达 │
│ │ │◄─────────────────│
│ │ │ │
│ │ 5. 检查脏数据 │ │
│ │ (flushed=0) │ │
│ │◄─────────────────┼──────────────────│
│ │ │ │
│ │ 6. 返回 PG 数据 │ │
│ │─────────────────────────────────────▶
│ │ (Parquet 被忽略 - 记录是脏的) │ACID 保证链
Forma 的一致性在三个关键点上依赖 PostgreSQL 的 ACID 属性。
写入路径,从应用到 PostgreSQL:
BEGIN;
INSERT INTO entity_main (...) VALUES (...);
INSERT INTO eav_data (...) VALUES (...);
INSERT INTO change_log (row_id, flushed_at) VALUES ($1, 0);
COMMIT; -- 全有或全无:三个插入要么全部成功,要么全部失败如果事务失败,不会有部分数据存在。记录要么完整存在(带有 flushed_at = 0),要么根本不存在。
导出路径,从 PostgreSQL 到 S3:
BEGIN;
-- 事务内读取是一致的
SELECT * FROM entity_main WHERE row_id IN (SELECT row_id FROM change_log WHERE flushed_at = 0);
-- 写入 S3(在事务外,但是幂等的)
-- ...S3 PUT...
-- 只有在 S3 确认后才标记为已刷新
UPDATE change_log SET flushed_at = now() WHERE row_id IN ($exported_ids) AND flushed_at = 0;
COMMIT;AND flushed_at = 0 子句是幂等性的来源,它防止 CDC 任务并发运行时的重复标记。
查询路径上,DuckDB 读取两个数据源,整体是一个时间点快照:读 change_log 获取 Dirty Set,读 Parquet(不可变文件,没有并发写入问题),读 PostgreSQL(使用快照隔离)。不需要锁,读者不会阻塞写者,写者也不会阻塞读者。
故障模式汇总
| 故障场景 | 数据状态 | 恢复动作 | 数据丢失? |
|---|---|---|---|
| 应用写入中途崩溃 | PG 事务回滚 | 自动(ACID) | 否 |
| CDC 在 S3 写入前崩溃 | 无变化 | 下次运行自动重试 | 否 |
| CDC 在 S3 写入后、PG 更新前崩溃 | S3 有数据,PG 显示"脏" | 查询从 PG 获取(正确);S3 文件成为孤儿 | 否 |
| S3 写入失败 | PG 无变化 | 下次运行自动重试 | 否 |
| DuckDB 查询失败 | 无副作用 | 客户端重试 | 否 |
| PostgreSQL 宕机 | 查询失败 | 退化到仅 Parquet 模式(降级模式) | 否* |
*在降级模式下,查询可能返回略微过时的数据,也就是上次成功导出的数据。如果"零脏读"是强制要求,Forma 也可以直接阻止查询。
优雅降级
当 DuckDB 或 S3 不可用时,Forma 不会崩溃,而是降级:
┌─────────────────────────────────────────────────────────────────────┐
│ 降级模式 │
├─────────────────────────────────────────────────────────────────────┤
│ 正常模式: PostgreSQL(热)+ DuckDB/Parquet(冷)→ 完整数据 │
│ S3 宕机: 仅 PostgreSQL → 仅热数据(最近的记录) │
│ PG 宕机: 仅 DuckDB/Parquet → 仅冷数据(可能过时) │
│ 两者都宕: 服务不可用(触发熔断器) │
└─────────────────────────────────────────────────────────────────────┘应用可以选择行为:严格模式下,任何数据源不可用就拒绝查询;尽力模式返回可用数据,附带警告头;缓存模式返回上次成功查询的缓存结果。
Last-Write-Wins:处理残留重复
Anti-Join 解决了 PostgreSQL 和 Parquet 之间的冲突。Parquet 内部还有一个问题。
由于 CDC 是增量导出,同一条记录可能在多个 Parquet 文件中存在多个版本:
delta/001.parquet:row_id=123, version=1delta/002.parquet:row_id=123, version=2
Forma 用 QUALIFY ROW_NUMBER() 实现 Last-Write-Wins:
SELECT *
FROM (
SELECT *,
ROW_NUMBER() OVER (PARTITION BY row_id ORDER BY updated_at DESC) AS rn
FROM read_parquet('s3://bucket/**/*.parquet')
)
WHERE rn = 1
AND (deleted_at IS NULL OR deleted_at = 0) -- 过滤软删除这样每个 row_id 只返回最新版本,已删除的记录也不会"复活"。
为什么选择 DuckDB?
为什么用 DuckDB 而不是 Trino、Spark,或者直接用 PostgreSQL 的 FDW?
| 特性 | DuckDB | Trino/Spark | PostgreSQL FDW |
|---|---|---|---|
| 部署复杂度 | 嵌入式,零部署 | 需要 3-10 节点集群 | 需要配置 FDW 扩展 |
| 冷启动延迟 | 50-100ms | 2-10 秒(JVM 预热) | 毫秒级(复用连接) |
| Parquet 原生支持 | 原生,向量化执行 | 好(需要 connector) | 需要 parquet_fdw 插件 |
| PostgreSQL 连接 | postgres_scanner | JDBC(额外延迟 10-50ms) | 内置 |
| 成本模型 | 按查询付费友好 | 集群常驻 $500-5000/月 | 依赖主库资源 |
DuckDB 是一个嵌入式 OLAP 引擎,可以直接嵌入到应用进程中,不需要额外的服务器。这对 Serverless 架构特别友好:Lambda 函数启动时加载 DuckDB(约 50MB),查询时直连 PostgreSQL(通过 postgres_scanner)和 S3(通过 httpfs),查询结束、Lambda 销毁,成本归零。
Serverless 成本模型
传统 OLAP 架构需要常驻集群,即使没有查询也在烧钱。DuckDB 嵌入式的特点让按需计费成为可能:
| 成本项 | 传统 OLAP 集群 | DuckDB Serverless |
|---|---|---|
| 空闲成本 | $500-5000/月 | $0 |
| 单次查询(1GB 扫描) | ~$0.001 | ~$0.005(含 Lambda) |
| 月均 1000 次查询 | $500-5000 | ~$5-10 |
对于查询量不大但数据量大的场景,比如历史审计、月度报表,Serverless 的成本优势可达 100-500 倍。
系列总结
回顾这三篇文章构建的完整架构:
┌─────────────────┐
│ Flexibility │
│ EAV + JSON │
│ Schema │
└────────┬────────┘
│
┌───────────────┼───────────────┐
│ │ │
▼ │ ▼
┌─────────────────┐ │ ┌─────────────────┐
│ Performance │ │ │ Cost │
│ Hot Table + │◀─────┴─────▶│ DuckDB + │
│ CTE JSON_AGG │ │ Serverless │
└─────────────────┘ └─────────────────┘三篇文章解决的问题
| 篇目 | 问题 | 解决方案 | 关键指标 |
|---|---|---|---|
| 第一篇 | Schema 灵活性 | EAV + JSON Schema + 热表 | 零 DDL,80/20 索引优化 |
| 第二篇 | N+1 查询 | CTE + JSON_AGG | 101→1 查询,1000ms→25ms |
| 第三篇 | 海量历史数据 | DuckDB + Anti-Join | 零脏读,Serverless 成本 |
核心设计原则
- 状态优先于时间戳。用
flushed_at明确标记同步状态,而不是比较时钟。 - 悲观优于乐观。宁可多查一次 PostgreSQL,也不冒读脏数据的风险。
- 计算下推。让 PostgreSQL 和 DuckDB 各司其职。
- 渐进降级。DuckDB 失败时回退到仅用 PG,保证可用性。
适用场景
这套架构适合数据结构频繁变化、需要 JSON Schema 灵活性的 AI 驱动应用;适合不同租户使用不同字段的多租户 SaaS,EAV 天然支持这一点;也适合历史数据聚合、报表、导出这类分析型查询,DuckDB + Parquet 处理起来很高效。
不适用场景
需要跨表 ACID 事务的场景,建议用纯 PostgreSQL。低于 10ms 的点查,建议 Redis 缓存加 PostgreSQL。实时流处理请用 Kafka 或 Flink,因为这里的 CDC 有分钟级延迟。
全栈对比:Forma vs. 替代方案
这套架构与其他常见的灵活可扩展数据存储方案相比:
| 能力 | Forma | MongoDB + Atlas | DynamoDB | PostgreSQL + TimescaleDB |
|---|---|---|---|---|
| Schema 灵活性 | ✅ JSON Schema | ✅ 无 Schema | ✅ 无 Schema | ⚠️ 需要 DDL |
| 跨记录 ACID | ✅ 完整 | ⚠️ 多文档受限 | ⚠️ 最多 25 项 | ✅ 完整 |
| SQL 兼容性 | ✅ 原生 | ❌ 仅 MQL | ❌ PartiQL(有限) | ✅ 原生 |
| 冷热分离 | ✅ 内置 | ⚠️ 手动分层 | ⚠️ 基于 TTL | ⚠️ 基于 chunk |
| Serverless 分析 | ✅ DuckDB | ⚠️ Atlas BI(贵) | ⚠️ Athena(独立) | ❌ 需要集群 |
| 冷存储成本 | ✅ S3 ($0.023/GB) | ⚠️ Atlas 归档 | ⚠️ 需要 S3 导出 | ⚠️ 基于磁盘 |
| 零脏读 | ✅ Dirty Set | ❌ 最终一致 | ⚠️ 单项强一致 | ✅ MVCC |
| AI 管道集成 | ✅ JSON Schema = LLM 契约 | ⚠️ 手动校验 | ⚠️ 手动校验 | ⚠️ 手动校验 |
Forma 的差异点在于组合:文档存储的灵活性加上关系型数据库的一致性,外加对 AI 工作负载的原生支持(JSON Schema 就是 LLM 与存储之间的契约),以及高性价比的冷数据分析(DuckDB + Parquet)。
结语
"Lakehouse"不是一个新概念,让人信任它才是难的部分。
Forma 的 Anti-Join + Dirty Set 机制是一个悲观的一致性协议:假设任何时候都可能有数据在路上,然后显式地处理这种不确定性。这比乐观的时间戳比较多了一些查询开销,需要扫描 change_log 表,换来的是可以证明的一致性保证。
在数据系统中,正确性排在性能前面,因为错误的快结果比正确的慢结果更糟。
系列导航
- [第一篇] 为什么 EAV 是 AI 时代最被低估的数据模型
- [第二篇] 杀死 N+1:一次 SQL 优化如何让延迟从 1 秒降到 25 毫秒
- [第三篇] 零脏读的 Serverless 湖仓:我们如何用 DuckDB 解决一致性难题 ← 当前(完结)
本文基于 Forma 项目的工程实践。Forma 是一个为 AI 时代设计的灵活数据存储引擎。
代码和讨论都在 GitHub 上。