供应链复杂业务实时数仓建设之路

背景

供应链业务是纷繁复杂的，我们既有 JIT 的现货模式中间夹着这大量的仓库作业环节，又有到仓的寄售，品牌业务，有非常复杂的逆向链路。在这么复杂的业务背后，我们需要精细化关注人货场车的效率和成本，每一单的及时履约情况，要做到这一点我们需要各粒度和维度的数据来支撑我们的精细化管理。

业务早期

业务早期，业务反馈我们后台管理系统某些报表查询慢。查询代码可知，如下图：

这种现象一般表现为：

大表 JOIN,rdbms 不擅长做数据聚合,查询响应慢,调优困难；
多表关联,索引优化,子查询优化,加剧了复杂度,大量索引,读库磁盘空间膨胀过快；
数据量大,多维分析困难,跨域取数,自助拉到实时数据困难等。

一方面原因是系统设计之初,我们主要关注业务流程功能设计,事务型业务流程数据建模,对于未来核心指标的落地,特别是关键实时指标落地在业务快速增长的情况下如何做到非常好的支撑。mysql 在此方面越来越捉襟见肘。

另外一方面原因是 mysql 这种 oltp 数据库是无法满足实时数据分析需求的,我们需要探索一套实时数据架构,拉通我们的履约,仓储,运配等各域的数据,做有效串联,因此我们开始了我们的实时数据架构探索，下图是我们一些思考。

附:数据视角的架构设计也是系统架构设计的重要组成部分。

架构演变

原始阶段

通过 Adb(AnalyticDB for MySQL)完成实时 join

通过阿里云 DTS 同步直接将业务库单表实时同步到 Adb，通过 Adb 强大的 join 能力和完全兼容 mysql 语法,可以执行任意 sql，对于单表大数据量场景或者单表和一些简单维表的 join 场景表现还是不错的,但是在业务复杂,复杂的 sql rt 很难满足要求,即使 rt 满足要求,单个 sql 所消耗的内存,cpu 也不尽人意,能支撑的并发量很有限。

通过 Otter 完成大宽表的建设

基于 Canal 开源产品，获取数据库增量日志数据并下发，下游消费增量数据直接生成大宽表，但是宽表还是写入 mysql 数据库，实现单表查询，单表查询速度显著提升，无 olap 数据库的常见做法，通过宽表减少 join 带来的性能消耗。

但是存在以下几个问题：

虽然 otter 有不错的封装,通过数据路由能做一些简单的数据拼接,但在调试上线复杂度上依然有不小的复杂度；
otter 伪装 mysql 从库同时要去做 etl 逻辑,把 cdc 干的活和实时 ETL 的活同时干了,耦合度较高。

实时架构 1.0

flink+kafka+ClickHouse

在上述调研尝试后都没有解决根本的问题，我们开始把目标建立标准的实时数仓的思路上来，在 20 年 olap 没有太多的可选项,我们把目标放在 clickhouse 上。

为了保证顺序 append 每次写入都会生成一个 part 文件，满足一定条件后台定时合并。
非常弱的 update delete，不能保证原子性和实时性。* clickhouse 只适合数据量大，业务模型简单,更新场景少的场景。
存算不分离，复杂查询影响 clickhouse 写入。

因为 clickhouse 的这些特性,尤其是不支持 upsert 的情况下,我们通常需要提前把大宽表的数据提前在 flink 聚合好,并且供应链数据生命周期长,作业流程也长如：

货物的生命周期较短时长为一周，长周期时长超过 1 个月；
库内环节异常的多，从卖家发货到收货、分拣、质检、拍照、鉴别、防伪、复查、打包、出库、买家签收等十几个甚至更多的环节，一张以商品实物 id 为主键的大宽表，需要 join 几十张业务表 ；
供应链系统早期设计没有每张表都会冗余唯一单号（入库单，作业单，履约单）这样的关键字段，导致没办法直接简单的 join 数据。
在这样一个架构下，们的 flink 在成本上，在稳定性维护上，调优上做的非常吃力。

附:clickhouse 不支持标准的 upsert 模式，可以通过使用 AggregatingMergeTree 引擎字段类型使用 SimpleAggregateFunction(anyLast, Nullable(UInt64)) 合并规则取最后一条非 null 数据可以实现 upsert 相似的功能，但读时合并性能有影响。

实时架构 2.0

flink+kafka+hologres

因此我们迫切的希望有支持 upsert 能力的 olap 数据库,同时能搞定供应链写多少的场景,也能搞定我们复杂查询的场景,我们希望的 olap 数据至少能做到如下几点：

有 upsert 能力，能对 flink 大任务做有效拆分；
存算分离，复杂业务计算，不影响业务写入，同时能平滑扩缩容；
有一定的 join 能力带来一些灵活度；
有完善的分区机制，热数据查询性能不受整体数据增长影响；
完善的数据备份机制。

这样一个行列混合的 olap 数据库,支持 upsert,支持存算分离,还是比较符合我们的预期。

目前这样一套架构支持了供应链每天数千人的报表取数需求,以及每天 10 亿数据量的导出,访问量在得物所有 to B 系统中排名靠前。

我们遇到的一些问题

多时间问题如何设置 segment_key,选择哪个业务字段作为 segment_key 供应链几十个环节都有操作时间,在不带 segment_key 的情况下性能如何保障，困扰了我们一段时间。

设置合理的 segment_key 如有序的时间字段,可以做到完全顺序写。每个 segment 文件都有个 min,max 值,所有的时间字段过来只需要去比较下在不在这个最小值最大值之间(这个动作开销很低),不在范围内直接跳过，在不带 segment_key 查询的条件下,也能极大的降低所需要过滤的文件数量。

批流融合背景：业务快速发展过程中，持续迭代实时任务成为常态。供应链业务复杂，环节多，流程往往长达一个月周期之久，这就导致 state ttl 设置周期长。job 的 operator 变化（sql 修改），checkpoint 无法自动恢复，savepoint 恢复机制无法满足，比如增加 group by 和 join。重新消费历史数据依赖上游 kafka 存储时效，kafka 在公司平台一般默认都是存储 7 天，不能满足一个月数据回刷需求场景。

方案：通过批流融合在 source 端实现离线 + 实时数据进行数据读取、补齐。

（1）离线按 key 去重，每个 key 只保留一条，减少消息量下发。（2）离线和实时数据合并，使用 last_value 取相同主键最新事件时间戳的一条数据。（3）使用 union all + group by 方式是可作为代替 join 的一个选择。（4）实时数据取当日数据，离线数据取历史数据，防止数据漂移，实时数据需前置一小时。

Join 算子乱序

问题分析

由于 join 算子是对 join 键做 hash 后走不同的分片处理数据 ，开启了 2 个并发后，再因为 header_id 字段的值变化，detail 表 2 次数据流走到了 2 个不同的 taskmanage,而不同的线程是无法保证输出有序性的 ，所以数据有一定的概率会乱序输出，导致期望的结果不正确，现象是数据丢失。

解决办法

通过 header inner join detail 表后，拿到 detail_id，这样再次通过 detail_id join 就不会出现（join 键）的值会从 null 变成非 null 的情况发生了，也就不会乱序了。

insert into sinkSelect detail.id,detail.header_id,header.idfrom detailleft join (    Select detail.id AS detail_id,detail.header_id,header.id    from header     inner join detail    on detail.header_id  =  header.id ) headerNewon detail.id  =  headerNew.detail_id