巨人肩膀
改版通知

巨人肩膀网站已全新改版。若您仍依赖旧站功能或数据，欢迎联系我们，我们会协助处理。联系我们

← 返回专栏列表

ApachePaimon在蚂蚁的生产实践

ckckck2025年1月10日44 浏览

Apache Paimon 中文社区官微

以下文章来源于 Apache Paimon，作者 闵文俊 & 朱育锋

Apache Paimon

Apache Paimon 是由 PMC 成员维护的中文社区官微。

背景

在蚂蚁数据研发的体系中，我们探索落地了一套基于 Flink 流批统一的计算架构。然而，由于存储系统的割裂，需要通过平台来做各种离线表、实时表的源表映射来统一源表的处理，用户理解成本和使用成本较高，难以推广和真正为用户减负。

因此，我们开始探索基于开源数据湖的方案来补足存储上的这一块拼图。我们调研了 Iceberg、Hudi 和 Paimon 数据湖格式，综合比较下来，选择了基于 Paimon 的数据湖方案。

Paimon 数据湖格式主要有以下特点：

  • 流批统一：支持批读批写、流读流写
  • 灵活的 Merge Engine：如 deduplicate、first_row、aggregation、partial-update
  • 灵活的 Change Log Producer：能够提供完整的流读 changelog
  • AppendOnly 表优化：支持小文件合并和索引优化
  • 分布式文件系统存储：低成本和可扩展的元数据

综合来看，Paimon 在流场景支持中做了很多工作，同时对传统的批处理场景也对标现有其他湖格式的能力。

业务场景

业务场景

总体架构如上，主要包含五个部分：数据入湖、元数据管理、离线计算入口、Flink SqlGateway 查询、流读入口。以下主要介绍两个实际使用数据湖优化开发体验和效率的场景。

直播用户特征序列

在支付宝的直播场景中，常见的一个需求是计算直播用户行为的序列特征。用户序列特征是指用户对于端上推荐内容做出的互动行为，包括曝光、点击、点赞等的汇总序列特征。

拆解到数据侧，需要产出一份 user x item 粒度的汇总宽表数据。在引入 Paimon 之前，采用的计算方案是所有聚合、打宽逻辑完全在 Flink 作业中实现，如下图所示：

直播用户特征序列

这个链路中主要有以下痛点：

  • 超大状态：由于聚合粒度为 user x item，状态中维护 Key 的数据量接近原始明细数据量级别，导致 Agg、Join 等算子的状态非常大
  • 性能较差：经常受上游流量冲高影响，使任务中超大状态算子成为性能瓶颈，导致任务反压延迟、吞吐下降，还会引发检查点失败等一系列问题
  • 探查成本高：中间结果（状态）维护在状态后端中，对外不可查，排查数据需另起链路，人力成本投入过高

因此，希望充分利用 Paimon 的 LSM Tree 机制，用户可以通过指定 Merge Engine，将原先任务中实现的聚合、打宽逻辑下沉到 Compaction 过程中，为实时链路降本增效。基于 Paimon 优化后的链路，如下图所示：

优化后的链路

可以看到在新链路中，Flink 任务变得更加轻量，原始链路中 Agg、Join 等超大状态算子都可以被省略。综合来看取得了以下收益：

  • 业务逻辑复杂度降低：无需在任务中手动声明聚合、打宽逻辑，因此可大幅简化业务逻辑、提升开发效率
  • 计算资源消耗降低：利用存储端合并期间的聚合能力，任务中状态存储与计算的开销大幅缩减，并且在流量高峰期，可牺牲一定的时效性来换取较高的吞吐
  • 探查成本降低：通过 Paimon 所提供的 audit_log 系统表，业务同学可以快速查询某一主键的中间变更过程，当出现数据问题时，可快速回溯，使排查效率提升 80% 左右

多表外键打宽

另一个在内部使用的场景是基于外键打宽的方案尝试。目前社区支持的还是两表基于同一主键的，通过 LSM tree 合并的方式来实现的两表 join。但是多表外键打宽在业务上实际上是很普遍的诉求。

传统离线场景中一般通过定期合并及多表关联实现大宽表，计算代价很高，时效性一般也是 T+1。同时我们在 Paimon 社区看到其他业务也在探索多表外键打宽的能力，因此我们基于社区讨论的一个方案，在 Paimon 上尝试解决这个问题。

以 a、b、c 三张表为例，其中主键分别为 pk1、pk2、pk3，每张表包含另一张表的外键，如何高效的获取这三张表的宽表呢？

  • a: pk1 pk2
  • b: pk2 pk3
  • c: pk3 xx

首先我们可以通过 Flink 流任务来实现这个需求，可以通过多流 Join 的方式，理想的情况下，这个流就可以产出预期的打宽数据了。但是 Flink Join 依赖本地状态存储，而且每两张表的 join 都会重复存储本地状态，表越多，状态存储大，性能越差。

sql 复制代码 
SELECT *
FROM a
INNER JOIN b ON a.pk2 = b.pk2
INNER JOIN c ON b.pk3 = c.pk3

在这个基础上，我们可以通过多流 join 只保留主键的部分，这样状态开销就会节省很多，性能也会提升。这样通过 Flink 流任务就可以以较低代价产出一张多表的主键变更关系流。

sql 复制代码 
INSERT INTO pk_relation_table
SELECT a.pk1, b.pk2, c.pk3
FROM a
INNER JOIN b ON a.pk2 = b.pk2
INNER JOIN c ON b.pk3 = c.pk3

然后下游流读这张 pk_relation_table，和原始的单表进行基于主键的维表关联，来补齐其他表的字段形成大宽表。

sql 复制代码 
SELECT *
FROM pk_relation_table AS x
JOIN a ON x.pk1 = a.pk1
JOIN b ON x.pk2 = b.pk2
JOIN c ON x.pk3 = c.pk3
多表外键打宽

整体链路如上，这样做的好处是：

  • 整个链路基于变更数据关联：因此只有增量数据触发变更，避免了周期性的全量关联
  • 流式触发关联：整体的时效性可以从 T+1 降低到分钟级
  • 原始数据只保留一份：不需要在关联的过程中在状态中重复存储

不过这里还有个缺陷，就是如果中间这个生成 pk 关系的 join 流任务重置，就会重新下发历史数据，那么消费这张 pk relation table 的下游表，又会回放之前的变更记录，导致下游全量重新 lookup 关联，这部分计算开销是完全无用的，而且会导致额外的开销。

这里的一个解决方法是用户可以在下游的任务中，通过过滤条件将历史已经完成关联的数据过滤，不过可能不是那么优雅。

针对这个问题，我们设计了另一种方案，来优化上面 pk relation table 的重置的情况。就是将中间的 pk relation table 也保存为一张 pk 表，并指定其 sequence 键。

sql 复制代码 
INSERT INTO pk_relation_table
SELECT pk1, pk2, pk3, sequence_1, sequence_2, sequence_3
FROM a
LEFT JOIN b ON a.pk2 = b.pk2
LEFT JOIN c ON b.pk3 = c.pk3

pk-relation-table 的 schema：

sql 复制代码 
CREATE TABLE PK_RELATION_TBALE (
  `pk1` varchar,
  `pk2` varchar,
  `pk3` varchar,
  `g_1` bigint,
  `g_2` bigint,
  `g_3` bigint,
  PRIMARY KEY (`pk1`, `pk2`, `pk3`) NOT ENFORCED
) WITH (
  'merge-engine' = 'partial-update',
  'fields.g_1.sequence_group' = 'g_1',
  'fields.g_2.sequence_group' = 'g_2',
  'fields.g_3.sequence_group' = 'g_3',
  'chaneglog-producer' = 'lookup'
)

其中 g_1、g_2、g_3 分别对应上游 3 张表的 gmt_modified 时间，即数据的版本。对于上游的写入，当任意一个流的版本变大时，才会发送 changelog 的数据，那么重置时也不会导致下游全部回刷重查了。相当于利用 Paimon sequence 防降的能力，当只有上游数据版本变大时才下发变更数据。

基于这个方案，可以实现多表的流式增量关联打宽，整体代价会远小于离线定期全量的打宽，数据新鲜度也高于离线的方案。目前还只是基于小数据量的表进行验证，因为整体的方案最终还依赖于 Paimon 的维表关联的性能，这块社区还在持续优化，我们也共同期待。

功能增强

在内部场景的支持中，我们也不断完善了 Paimon 的很多功能。主要有：

  • 支持 session 级别动态参数设置：支持通过 set 语句设置表级别的动态参数覆盖 catalog 中的参数，实现在即席查询中参数动态生效的能力
  • 优化查询分区下推能力：优化原始的 query filter 下推，支持分区条件的完全下推，避免分区 limit query 扫描全分区的问题
  • 支持 Partition lookup join：支持 hash 分区的 Partition join 策略，可以大大减小 Paimon 维表关联时需要缓存的数据量，降低存储开销
  • 支持 partial insert union sink：优化用户多表写入任务的体验，只需要通过多个 partial insert 语句，然后由 planner 自动优化插入 Union 节点，合并多个 sink 算子。避免用户手动 union 以及 Cast null 的操作
  • catalog 支持多存储后端设计：内部使用的 catalog 会支持多个存储后端，如基于 hdd 的 dfs、基于 ssd 的 dfs 以及 oss 等等，因此我们设计了一个可以支持多存储后端的 catalog，通过 database 级别的参数动态获取存储后端的配置，提升了管理上的灵活性
  • 支持 first_row merge engine：支持 first_row 模式的 merge engine，可以对 append 流进行去重处理，类似于 Flink 中的 first row 去重，其好处是通过一个去重表之后，下游可以多个任务复用，避免重复的去重开销
  • 支持 partial-update with aggregation：支持多表合并的时候指定字段的聚合策略，在 1 对多合并的场景中，通过指定聚合策略，能更灵活的满足业务上的诉求

以上绝大部分功能已贡献给社区，也非常感谢 Paimon 社区对这些功能的接纳与建议。

总结与展望

经过半年的探索，内部基于 Paimon 的数据湖产品基本完成落地，开始服务部分内部业务。在此也非常感谢 Paimon 社区的支持，后续我们将继续针对内部业务场景进行优化，主要方向包括：

  • 跟进社区维表关联的优化：提升外键打宽的适用范围
  • 搭建数据湖管理平台和表管理服务
  • 探索数据湖上索引的应用：优化查询性能
总结与展望

end