1、幸福里基于 Flink&Paimon 的流式数仓实践李国君字节跳动基础架构工程师Streaming lakehouse meetup背景0101CONTENT目录 0202流式数仓实践0303调优0404未来规划01背景业务背景以工单系统工单系统为例：1.经纪人完成一次带看，在系统提交工单2.门店经理审核工单3.以上两条数据更新到业务库，业务库的 Binlog 数据进入实时数仓4.数仓结果产生数据报表，或用于考核系统实时数仓建模特点准确性要求准确性要求 100%100%与考核，奖励挂钩。要求数据不丢不重全量计算全量计算增量数据在 MQ 留存时间有限，需要拿到全量数据 view 进行计算业务实时

2、数仓 Pipeline准确性要求准确性要求 100%100%：状态维护全量数据全量计算：全量计算：每一层消费MQ MQ 实时增量数据+Hive Hive 离线全量数据01 01 开发复杂度高开发复杂度高痛点分析-开发复杂度高select col1,col2,.FROM (select*,row_number()over(partition by someKey order by biz_time DESC,nanotime DESC)as rn from table AS t1)as t1where rn=1select col1,col2,nanoTime()as ts from table

3、代价代价需要额外额外处理数据重复重复引入非确定性计算非确定性计算原因原因存储异构存储异构，Base+Delta 数据难对齐代价代价数据排查排查&验证验证需要额外 dump hive数据订正订正难原因原因数据订正产生结果回退结果回退暴露给用户血缘关系血缘关系复杂且由人为维护02 02 运维成本高运维成本高痛点分析-运维成本高-代价代价存储、计算资源开销资源开销大作业恢复慢恢复慢原因原因去重算子维护全量数据状态全量数据状态级联 Join 状态重复状态重复03 03 状态大状态大痛点分析-状态大贡献 30%状态量为什么选择Paimon原始链路问题1.1.存储异构，存储异构，Base+Delta 数据

4、难对齐2.去重引入非确定性计算非确定性计算和大状态大状态3.3.血缘关系血缘关系复杂&数据订正结果回退结果回退暴露给用户Paimon1.1.流批一体存储流批一体存储2.2.不需要去重不需要去重，changelog producer 代替状态算子3.3.血缘管理血缘管理&数据一致性管理数据一致性管理，支持无感知数据订正02 流式数仓实践架构设计流批一体存储流批一体存储统一的 Table 抽象支持高效的流/批访问流批一体计算引擎流批一体计算引擎用 Flink 统一 Streaming/Batch/MPP 计算血缘信息管理血缘信息管理为数据质量管理提供基础数据致性管理流批体 ETL 数据管理业务流式

5、数仓Pipeline流批一体存储流批一体存储流批一体计算流批一体计算中间结果可查中间结果可查收益简化开发流程流批一体存储解决实时离线存储异构减少业务入侵移除去重算子解决非确定性计算减少状态量Changelog 持久化减少 30%状态量提升运维体验中间数据可查血缘关系&多表一致性增强多表关联调试能力数据订正无感知数据可追溯问题45数据新鲜度差小文件问题端到端延迟：秒级-分钟级影响读写性能4503 调优端到端延迟调优-数据可见性&Checkpoint结论：数据可见性与 Checkpoint 绑定Checkpoint 周期=Checkpoint interval+Checkpoint latency

6、Paimon 流写 Pipeline1.Writer 生成数据文件，但不对外可见2.Checkpoint commit 阶段：Commiter 生成 snapshot，数据对外可见解决非确定性计算解决非确定性计算端到端延迟调优-调小Checkpoint Interval过于频繁的 Checkpoint 会影响写入性能小文件Compaction 代价变高结论：Checkpoint interval 设置为 1-2min 比较合适4545端到端延迟调优-Checkpoint Latency 优化Log-Based 增量 Checkpoint减小状态量Checkpoint 持续上传搭建独立 HDFS

7、 集群减少遇到慢节点概率持续上传本地状态文件持续上传 Checkpoint 的增量 changelog减少上传数据量结论：端到端延迟 1min+解决非确定性计算解决非确定性计算小文件问题小文件小文件明显小于 HDFS 上 block 大小的文件文件数量多-更多 block-更多更多 IOIO影响影响稳定性变差降低查询效率资源浪费解决非确定性计算解决非确定性计算小文件优化影响因素影响因素文件生成文件生成Checkpoint intervalWriteBuffer 大小文件划分文件划分数据量大小Key 的设置:热 key 问题文件清理文件清理Compaction 触发条件Snapshot 过期条件

8、-解决非确定性计算解决非确定性计算小文件优化影响因素影响因素参数属性参数属性参数参数推荐设置推荐设置Checkpoint intervalFlinkexecution.checkpointing.interval1-2minWriteBuffer 大小Paimonwrite-buffer-size使用默认值数据量Paimonbucket单个 Bucket 文件大小在 1G 左右比较合适热 key 问题Paimonbucket-keypartition根据业务数据设置Compaction 管理Paimonnum-sorted-paction-triggernum-sorted-run.stop-

9、trigger使用默认值Snapshot 管理Paimonsnapshot.num-retained.maxsnapshot.num-retained.minsnapshot.time-retained使用默认值解决非确定性计算解决非确定性计算整体指标对比原有实时数仓链路原有实时数仓链路流式数仓链路流式数仓链路端到端延迟端到端延迟1 1-2min(2min(开开 minimini-batch)batch)1 1-2min2min数据排查时效性数据排查时效性小时级小时级分钟级分钟级状态量状态量节省约节省约 30%30%开发周期开发周期缩短约缩短约 50%50%04 未来规划Embeded Log SystemEmbeded Log System数据可见性与数据可见性与 Checkpoint Checkpoint 解绑解绑秒级端到端延迟秒级端到端延迟血缘关系管理血缘关系管理数据订正能力增强数据订正能力增强Join Join 状态复用状态复用中间状态可查中间状态可查数据一致性管理数据一致性管理状态复用状态复用未来规划建立监控建立监控体系体系指标可观测指标可观测监控运维监控运维ThanksStreaming lakehouse meetup