个人介绍

李昂
高级数据研发工程师
Apache Doris & Hudi Contributor

业务背景

部门成立早期, 为了应对业务的快速增长, 数仓架构采用了最直接的Lambda架构

1. 对数据新鲜度要求不高的数据, 采用离线数仓做维度建模, 采用每小时调度binlog+每日主键归并的方式实现T+1数据更新
2. 对数据时效性要求比较高的业务, 采用实时架构, 保证增量数据即时更新能力, 另一方面, 为了保证整体上线效率, 存量数据采用离线SQL处理, 以提高计算吞吐量

Lambda整体架构如下

此时的架构存在以下缺陷

1. **逻辑冗余 : **同一个业务方案, 往往有离线与实时两套开发逻辑, 代码复用性低, 需求迭代成本大, 任务交接、项目管理复杂
2. **数据不一致 : **应用层数据来源有多条链路, 在处理逻辑异构的情况下, 存在数据不一致的问题, 且问题排查成本大, 周期长
3. **数据孤岛 : **随着业务增长, 为了应对离线批处理、OLAP分析、C端高并发点查等场景, 引入的存储引擎越来越多, 存在数据孤岛

基于上述Lambda架构存在的缺陷, 我们希望对其作出改进, 实现以下目的

1. 流批一体 : 同一个业务方案, 可以由一套代码逻辑或者核心逻辑一致的SQL实现
2. 数据整合 : 统一离线批处理与OLAP分析的数据存储口径, 同时查询支持SparkSQL与Doris-Multi-Catalog, 打破数据孤岛

选型调研

综合考虑以下几点

1. 项目成熟度 : 社区活跃度、国内Committer数量、国内群聊活跃度、各公司最佳实践发文等
2. 数据初始化能力 : 考虑到需要对历史项目进行覆盖, 需要考虑存量数据写入能力
3. 数据更新能力 : 一方面是数据根据PrimaryKey或者UniqueKey的实时Upsert、Delete性能, 另一方面是Compaction性能
4. CDC : 如果需要分层处理, 则要求数据湖作为Flink Source时有产生撤回流的能力

我们最终选定使用Apache Hudi作为数据湖底座

方案选型

业务痛点

实时流 join 是事实数仓的痛点之一, 在我们的场景下, 一条事实数据, 需要与多个维度的数据做关联, 例如一场司法拍卖, 需要关联企业最新名称、董监高、企业性质、上市信息、委托法院、询价评估机构等多个维度;一方面, 公司与董监高是1:N的对应关系, 无法实现一条写入, 多条更新; 另一方面, 企业最新名称的变更, 可能涉及到历史冷数据的更新

方案设计

FlinkSQL+离线修复

方案描述
通过FlinkSQL实现增量数据的计算, 每日因为状态TTL过期或者lookup表变更而没有被命中的数据, 通过凌晨的离线调度进行修复

优点

1. SQL开发 : 便于维护
2. 架构简洁 : 不涉及其他非必要组件

缺点

1. **批流没有完全一体 : **同一逻辑仍然并存FlinkSQL与SparkSQL两种执行方式
2. **维护Flink大状态 : **为保证数据尽可能的join到, 需要设置天级甚至周级的TTL
3. **时效性下限较低 : **最差仍然可能存在T+1的延迟

MySQL中间表

方案描述
使用MySQL实现数仓分层, 为每张上游表, 都开发lookup逻辑, Hudi只负责做MySQL表的镜像

优点

1. 真正流批一体 : 整个链路彻底摆脱离线逻辑
2. **时效性最高 : **所有更新都能及时反映到下游

缺点

1. 维护成本大 : 每张Hudi表都镜像于一张MySQL表, 链路加长, 复杂度提高
2. 存储冗余 : 每张表各在MySQL与Hudi存一份, 同时, lookup还需要索引支撑, 磁盘占用高

最终结论

1. 因为C端业务的特殊性, 需要MySQL提供点查能力, 所以第二种方案的磁盘冗余处于可接受范围
2. 第一种方案T+1的下限无法被接受, 若提高离线修复的频率, 考虑到Flink已经维护大状态, 或将需要较大的内存开销

所以最终方案选定为第二种 : **MySQL中间表方案**, 优化后的整体架构如下

1. ODS层的Hudi充当一个Queryable Kafka, 提供CDC给下游数据
2. 实时ETL通过MySQL完成, 对于每一张新的结果表, 都会原样镜像一份到Hudi
3. Doris与Hive通过读RO表完成与Hudi的统一集成

方案实施

增量实时写入

table.type

根据上述方案, 我们的数据写入是完全镜像于每个flink job的产出MySQL表, 绝大部分表日更新量在50w~300w, 为了保证写入的稳定性, 我们决定采用MOR表

index.type

在选择index的时候, 因为BLOOM随着数据量的上升, 瓶颈出现比较快, 我们的候选方式有FLINK_STATE与BUCKET, 综合考虑以下几点要素

1. **数据量 : **当数据量超过5e, 社区的推荐方案是使用BUCKET, 目前我们常见的表数据量浮动在2e - 4e
2. **维护成本 : **使用Flink_STATE作为index时, 程序重启如果没有从检查点恢复, 需要开始bootstrap重新加载索引
3. **资源占用 : **为保证稳定, FLINK_STATE需要TaskManager划分0.5~1G左右的内存用于运行Rocksdb, 而BUCKET则几乎不需要状态开销
4. **横向扩展 : **bucket_num一经确认, 则无法更改(高版本的CONSISTENT_HASHING BUCKET依赖Clustering可以实现动态bucket_num), FLINK_STATE无相关概念

我们最终选择使用BUCKET

1. 它不与RocksDB绑定, 资源占用较低
2. 不需要bootstrap, 便于维护
3. 考虑到数据量与横向扩展, 我们预估数据量为5e~10e, 在该场景下, BUCKET会有更好的表现

同时, 综合参考社区推荐与相关最佳实践的文献,

1. 我们限制每个Parquet文件在2G以内
2. 假设Parquet+Gzip的压缩比率在5:1
3. 预估数据量在10e量级的表

最终, 我们设置bucket_num为128

离线写入

为了快速整合到历史已经上线的表, 存量数据的快速导入同样也是必不可少的, 通过官网学习, 我们设计了两种方案

1. **bulk_insert : **优点是速度快, 没有log小文件, 缺点是不够便捷, 需要学习和引入成本
2. **大并发的upsert : **优点是只需要加大并行度, 使用最简单, 缺点是产生大量小文件, 写入完毕后第一次compaction非常耗费资源

在分别对上述2种方案进行测试后, 我们决定采用bulk_insert的方式, 最大的因素还是大并发的Upsert在第一次写入后, 需要的compaction资源非常大, 需要在第一次compaction后再次调整运行资源, 不便于自动化

Compaction

• 同步
  • **优点 : **便于维护
  • **缺点 : **流量比较大的时候, 干扰写流程; 在存量数据大, 增量数据小的情况下, 资源难以分配
• 异步
  • **优点 : **与同步任务隔离, 不干扰写流程, 可*配置资源
  • **缺点 : **对于每个表, 都需要单独维护一个定时任务

综合考虑运维难度与资源分配后, 我们决定采用异步调度的方式, 因为我们读的都是RO表, 所以对Compaction频率和单次Compaction时间都有限制, 目前的方案是Compacion Plan由同步任务生成, Checkpoint Interval为1分钟, 触发策略为15次Commits

成果落地

流批一体

整合实时链路与离线链路, 所有产出表均由实时逻辑产出

• 开发工时由之前普遍的离线2PD+实时3PD提升至实时3PD, 效率提升40%
• 每个单元维护成本由1名实时组同学+1名离线组同学变更为只需要1位实时组同学, 维护成本节约50%

数据整合

配合Doris多源Catalog, 完成数据整合, 打破数据孤岛

• 使司内Doris集群完成存算分离与读写分离, 节约磁盘资源30T+, 集群故障率由最高3次/月降至1次/月, 稳定性提升70%
• 下线高性能(SSD存储)Hbase与GaussDB, 节约成本50w/年

平衡计算压力

之前Hive的每日数据由单独离线集群通过凌晨的多路归并完成多版本合并, 目前只需要一个实时集群

• 退订离线集群70%弹性节点, 节约成本30w/年

经验总结

Checkpoint反压优化

在我们测试写入的时候, Checkpoint时间比较长, 而且会有反压产生, 追踪StreamWriteFunction.processElement()方法, 发现数据缓情况如下

为了将flush的压力分摊开, 我们的方案就是减小buffer

ps : 默认write.task.max.size必须大于228M
最终的参数 :

-- index
'index.type' = 'BUCKET',
'hoodie.bucket.index.num.buckets' = '128',
-- write
'write.tasks' = '4',
'write.task.max.size' = '512',
'write.batch.size' = '8',
'write.log_block.size' = '64',

FlinkSQL TIMESTAMP类型兼容性

当表结构中有TIMESTAMP(0)数据类型时, 在使用bulk_insert写入存量数据后, 对接upsert流并进行compaction时, 会报错

Caused by: java.lang.IllegalArgumentException: INT96 is deprecated. As interim enable READ_INT96_AS_FIXED flag to read as byte array.

提交issue https://github.com/apache/hudi/issues/9804 与社区沟通
最终发现是TIMESTAMP类型, 目前只对TIMESTAMP(3)与TIMESTAMP(6)进行了parquet文件与avro文件的类型标准化
解决方法是暂时使用TIMESTAMP(3)替代TIMESTAMP(0)

Hudi Hive Sync Fail

将Hudi表信息同步到Hive原数据时, 遇到报错, 且无法通过修改pom文件依赖解决

java.lang.NoSuchMethodError: org.apache.parquet.schema.Types$PrimitiveBuilder.as(Lorg/apache/parquet/schema/LogicalTypeAnnotation;)Lorg/apache/parquet/schema/Types$Builder

与社区沟通, 发现了相同的问题 https://github.com/apache/hudi/issues/3042
解决方法是修改源码的 packaging/hudi-flink-bundle/pom.xml , 加入

<relocation>
  <pattern>org.apache.parquet</pattern>
  <shadedPattern>${flink.bundle.shade.prefix}org.apache.parquet</shadedPattern>
</relocation>

并使用

mvn clean install package -Dflink1.17 -Dscala2.12 -DskipTests -Drat.skip=true -Pflink-bundle-shade-hive3 -T 10

手动install源码, 在程序的pom文件中, 使用自己编译的jar包

Hudi Hive Sync 使用 UTC 时区

当使用FlinkSQL TIMESTAMP(3)数据类型写入Hudi, 并开启Hive Sync的时, 查询Hive中的数据, timestamp类型总是比原值多8小时
原因是Hudi写入数据时, 支持UTC时区, 详情见issue https://github.com/apache/hudi/issues/9424
目前的解决方法是写入数据时, 使用FlinkSQL的 CONVERT_TZ 函数

insert into dwd
select
id,CAST(CONVERT_TZ(CAST(op_ts AS STRING), 'Asia/Shanghai', 'UTC') AS TIMESTAMP(3)) op_ts
from ods;

HoodieConfig.setDefaults() NPE

在TaskManager初始化阶段, 偶尔遇到NPE, 且调用栈如下

java.lang.NullPointerException: null
at org.apache.hudi.common.config.HoodieConfig.setDefaults(HoodieConfig.java:123)

通过与社区交流, 发现是ReflectionUtils的CLAZZ_CACHE使用HashMap存在线程安全问题
解决方法是引入社区提供的PR : https://github.com/apache/hudi/pull/9786 通过ConcurrentHashMap解除线程安全问题

未来规划

Metric监控

对接Pushgateway、Prometheus与Grafana, 通过图形化更直截了当的监控Hudi内部相关服务、进程的内存与CPU占用情况, 做到

1. 优化资源, 提升程序稳定性
2. 排查潜在不确定因素, 风险预判
3. 接入告警, 加速问题响应

统一元数据管理

目前是采用封装工具类的方式, 让每个开发同学在产出一张结果表的同时, 在同一个job中启动一条Hudi同步链路, 缺少对Hudi同步任务的统一管理与把控, 后续准备对所有Hudi链路迁出, 进行统一的任务整合与元数据管理

引入CONSISTENT_HASHING BUCKET

后续计划中我们希望在1.0发行版中可以正式将CONSISTENT_HASHING BUCKET投入到线上环境, 现在线上许多3e~5e量级的表都是提前按照10e数据量来预估资源与bucket_num, 有资源浪费的情况, 希望可以通过引入一致性hash的bucket索引, 来解决上述问题