关于jquery:菜鸟实时数仓20进阶之路

简介： 供应链物流场景下的业务复杂度高，业务链路长，节点多，实体多，实时数仓建设难度高。菜鸟跨境进口业务场景更是如此，更简单的场景带来更简单的实体数据模型，对接的业务零碎多导致 ETL 流程特地简单，还有海量的日均解决数据量，使得团队在建设进口实时数仓的过程中，面临着诸多挑战。

如何保障简单实体关系下的数据准确性？如何升高多数据源状况下的数据处理复杂度？如何晋升实时多流 Join 的解决效率？如何实现实时超时统计？如何实现异常情况下的数据状态复原？本文次要分享菜鸟进口实时数仓的降级教训，以及如何利用 Flink 的个性解决在开发实际中遇到的问题。

次要内容包含：

1. 相干背景介绍
2. 进口实时数仓演进过程
3. 挑战及实际
4. 总结与瞻望

01 相干背景介绍

1. 进口业务简介

进口业务的流程大抵比拟清晰，国内的买家下单之后，国外的卖家发货，通过清关，支线运输，到国内的清关，配送，到消费者手里，菜鸟在整个过程中负责协调链路上的各个资源，实现物流履约的服务。去年考拉融入到阿里体系之后，整个进口业务规模占国内进口单量的规模是十分高的。并且每年的单量都在迅速增长，订单履行周期特地长，两头波及的环节多，所以在数据建设时，既要思考把所有数据交融到一起，还要保证数据有效性，是十分艰难的一件事件。

2. 实时数仓加工流程

① 个别过程

上面简略介绍一下实时数仓的加工流程，个别会对接业务库或者日志源，通过数据同步的形式，比方 Sqoop 或 DataX 把音讯同步到消息中间件中暂存，上游会接一个实时计算引擎，对音讯进行生产，生产之后会进行计算、加工，产出一些明细表或汇总指标，放到查问服务上供数据利用端应用。

② 菜鸟外部流程

在菜鸟外部也是同样的流程，咱们将业务库数据通过 DRC (数据备份核心) 增量采集 Binlog 日志的形式，同步到 TT (相似 Kafka 的消息中间件) 做一个音讯暂存，前面会接一个 Flink 实时计算引擎进行生产，计算好之后写入两种查问服务，一种是 ADB，一种是 HBase (Lindorm)，ADB 是一个 OLAP 引擎，阿里云对外也提供服务，次要是提供一些丰盛的多维分析查问，写入的也是一些维度比拟丰盛的轻度汇总或明细数据，对于实时大屏的场景，因为维度比拟少，指标比拟固定，咱们会积淀一些高度汇总指标写到 HBase 中供实时大屏应用。

02 进口实时数仓演进过程

接下来讲一下进口实时数仓的演进过程：

2014 年：进口业务线大略在 14 年时，建好了离线数仓，能提供日报。

2015 年：能提供小时报，更新频度从天到小时。

2016 年：基于 JStorm 摸索了一些实时指标的计算服务，越来越趋向于实时化。因为 16 年刚开始尝试实时指标，指标还不是特地丰盛。

2017 年：菜鸟引进了 Blink，也就是 Flink 在阿里的外部版本，作为咱们的流计算引擎，并且进口业务线在同一年买通了实时明细，通过实时明细大宽表对外提供数据服务。

2018 年：实现了菜鸟进口实时数仓 1.0 的建设。

2020 年：开始了实时数仓 2.0 的建设，为什么开始 2.0？因为 1.0 在设计过程中存在了很多问题，整个模型架构不够灵便，扩展性不高，还有一些是因为没有理解 Blink 的个性，导致误用带来的一些运维老本的减少，所以前面进行了大的降级革新。

1. 实时数仓 1.0

接下来讲一下实时数仓 1.0 的状况，一开始因为在倒退初期，业务模式不太稳固，所以一开始的策略就是围绕业务小步快跑，比方针对业务 1 会开发一套实时明细层，针对业务 2 也会开发一套实时工作，益处是能够随着业务倒退疾速迭代，相互之间不影响，晚期会更灵便。

如上图右侧所示，最底层是各个业务零碎的音讯源，实时工作次要有两层，一层是实时明细层，针对业务线会开发不同的明细表，明细表就是针对该条业务线须要的数据把它抽取过去，在这之上是 ADM 层，也就是实时应用层，应用层次要针对具体的场景定制，比方有个场景要看整体汇总指标，则从各个明细表抽取数据，产生一张实时汇总层表，整个过程是竖向烟囱式开发，模型比拟凌乱，难扩大，并且存在很多反复计算。

前面也是因为反复计算的问题，进行了一层形象，加了一个前置中间层，对公共的局部进行提取，然而治标不治本，整个模型还是比拟凌乱的，数据建设上也没有进行对立，模型扩展性上也很差。

2. 实时数仓 2.0

2.0 降级完之后是比拟清晰的一张图：

前置层：底层数据源会接入到前置中间层，屏蔽掉底层一些非常复杂的逻辑。

• 明细层：前置层会把比拟洁净的数据给到明细表，明细层买通了各个业务线，进行了模型的对立。
• 汇总层：明细层之上会有轻度汇总和高度汇总，轻度汇总表维度十分多，次要写入到 OLAP 引擎中供多维查问剖析，高度汇总指标次要针对实时大屏场景进行积淀。
• 接口服务：汇总层之上会依据对立的接口服务对外提供数据输入。
• 数据利用：应用层次要接入包含实时大屏，数据利用，实时报表以及音讯推送等。

这就是实时数仓 2.0 降级之后的模型，整个模型尽管看起来比较简单，其实背地从模型设计到开发落地，遇到了很多艰难，破费了很大的精力。上面为大家分享下咱们在降级过程中遇到的挑战及实际。

03 挑战及实际

咱们在实时数仓降级的过程中，面临的挑战如下：

1. 业务线和业务模式多

第一个就是对接的业务线比拟多，不同的业务线有不同的模式，导致一开始小步快跑形式的模型比拟割裂，模型和模型之间没有复用性，开发和运维老本都很高，资源耗费重大。

解决方案：逻辑中间层降级

咱们想到的比较简单的思路就是建设对立的数据中间层，比方业务 A 有出库、揽收、派送等几个业务节点，业务 B 可能是另外几个节点，整个模型是割裂的状态，但实际上业务倒退到中后期比较稳定的时候，各个业务模式之间绝对比较稳定，这个时候能够对数据进行一个形象，比方业务 A 有节点 1、节点 5 和其余几个业务模式是一样的，通过这种对齐的形式，找出哪些是公共的，哪些是非公共的，提取进去积淀到逻辑中间层里，从而屏蔽各业务之间的差距，实现对立的数据建设。把逻辑中间层进行对立，还有一个很大的起因，业务 A，B，C 尽管是不同的业务零碎，比方履行零碎，关务零碎，然而实质上都是同一套，底层数据源也是进行各种形象，所以数仓建模上也要通过对立的思路进行建设。

2. 业务零碎多，超大数据源

第二个就是对接的零碎十分多，每个零碎数据量很大，每天亿级别的数据源就有十几个，梳理起来十分艰难。带来的问题也比拟显著，第一个问题就是大状态的问题，须要在 Flink 里保护特地大的状态，还有就是接入这么多数据源之后，老本怎么管制。

解决方案：善用 State

State 是 Flink 的一大个性，因为它能力保障状态计算，须要更正当的利用。咱们要认清 State 是干什么的，什么时候须要 State，如何优化它，这些都是须要思考的事件。State 有两种，一种是 KeyedState，具体是跟数据的 Key 相干的，例如 SQL 中的 Group By，Flink 会依照键值进行相干数据的存储，比方存储到二进制的一个数组里。第二个是 OperatorState，跟具体的算子相干，比方用来记录 Source Connector 里读取的 Offset，或者算子之间工作 Failover 之后，状态怎么在不同算子之间进行复原。

① 数据接入时 ” 去重 ”

上面举个例子，怎么用到 KeyedState，比方物流订单流和履行日志流，两个作业关联产生出最终须要的一张大表，Join 是怎么存储的呢？流是始终不停的过去的，音讯达到的前后程序可能不统一，须要把它存在算子外面，对于 Join 的状态节点，比较简单粗犷的形式是把左流和右流同时存下来，通过这样的形式保障不论音讯是先到还是后到，至多保障算子外面数据是全的，哪怕其中一个流很晚才达到，也能保障匹配到之前的数据，须要留神的一点是，State 存储依据上游不同而不同，比方在上游定义了一个主键 Rowkey，并且 JoinKey 蕴含了主键，就不存在多笔订单对应同一个外键，这样就通知 State 只须要依照 JoinKey 存储惟一行就能够了。如果上游有主键，然而 JoinKey 不蕴含 Rowkey 的话，就须要在 State 里将两个 Rowkey 的订单同时存下来。最差的状况是，上游没有主键，比方同一笔订单有 10 条音讯，会有先后顺序，最初一条是无效的，然而对于零碎来说不晓得哪条是无效的，没有指定主键也不好去重，它就会全副存下来，特地耗资源和性能，相对来说是特地差的一种形式。

因而，咱们在数据接入时进行 ” 去重 ”。数据接入时，依照 row_number 进行排序，通知零碎依照主键进行数据更新就能够了，解决 10 条音讯不晓得应该存几条的问题。在下面这个 case 外面，就是依照主键进行更新，每次取最初一条音讯。

依照 row_number 这种形式并不会缩小数据处理量，然而会大大减少 State 存储量，每一个 State 只存一份无效的状态，而不是把它所有的历史数据都记录下来。

② 多流 join 优化

第二个是多流 Join 的优化，比方像上图左侧的伪代码，一张主表关联很多数据源产生一个明细大宽表，这是咱们喜爱的形式，然而这样并不好，为什么呢？这样一个 SQL 在实时计算里会依照双流 Join 的形式顺次解决，每次只能解决一个 Join，所以像右边这个代码里有 10 个 Join，在左边就会有 10 个 Join 节点，Join 节点会同时将左流和右流的数据全副存下来，所以会看到左边这个图的红框里，每一个 Join 节点会同时存储左流和右流的节点，假如咱们订单源有 1 亿，外面存的就是 10 亿，这个数据量存储是十分可怕的。

另外一个就是链路特地长，不停的要进行网络传输，计算，工作提早也是很大的。像十几个数据源取数关联在一起，在咱们的理论场景是实在存在的，而且咱们的关联关系比这个还要更简单。

那咱们怎么优化呢？咱们采纳 Union All 的形式，把数据错位拼接到一起，前面加一层 Group By，相当于将 Join 关联转换成 Group By，它的执行图就像上图右侧这样，黄色是数据接入过程中须要进行的存储，红色是一个 Join 节点，所以整个过程须要存储的 State 是非常少的，主表会在黄色框和红色框别离存一份，别看数据源十分多，其实只会存一份数据，比方咱们的物流订单是 1000 万，其余数据源也是 1000 万，最终的后果无效行就是 1000 万，数据存储量其实是不高的，假如又新接了数据源，可能又是 1000 万的日志量，但其实无效记录就是 1000 万，只是减少了一个数据源，进行了一个数据更新，新增数据源老本近乎为 0，所以用 Union All 替换 Join 的形式在 State 里是一个大大的优化。

1. 取数外键多，易乱序

第三个是取数外键多，乱序的问题，乱序其实有很多种，采集零碎采集过去就是乱序的，或者传输过程中导致的乱序，咱们这边要探讨的是，在理论开发过程中不小心导致的乱序，因为其余层面的货色平台曾经帮咱们思考好了，提供了很好的端到端的一致性保障。

举个例子比如说有两个单子都是物流单，依据单号取一些仓内的音讯，音讯 1 和音讯 2 先后进入流解决外面，关联的时候依据 JoinKey 进行 Shuffle，在这种状况下，两个音讯会流到不同的算子并发上，如果这两个并发处理速度不统一，就有可能导致先进入零碎的音讯后实现解决，比方音讯 1 先达到零碎的，然而解决比较慢，音讯 2 反倒先产出，导致最终的输入后果是不对的，实质上是多并发场景下，数据处理流向的不确定性，同一笔订单的多笔音讯流到不同的中央进行计算，就可能会导致乱序。

所以，同一笔订单音讯解决完之后，如何保障是有序的？

上图是一个简化的过程，业务库流入到 Kafka，Binlog 日志是程序写入的，须要采纳肯定的策略，也是程序采集，能够依据主键进行 Hash 分区，写到 Kafka 外面，保障 Kafka 外面每个分区存的数据是同一个 Key，首先在这个层面保障有序。而后 Flink 生产 Kafka 时，须要设置正当的并发，保障一个分区的数据由一个 Operator 负责，如果一个分区由两个 Operator 负责，就会存在相似于方才的状况，导致音讯乱序。另外还要配合上游的利用，能保障依照某些主键进行更新或删除操作，这样能力保障端到端的一致性。

Flink 曾经配合上下游零碎曾经帮咱们实现了端到端的一致性性能，咱们只须要保障外部解决工作不能乱序。咱们的解法是防止 Join Key 发生变化，如提前通过非凡映射关系把 Join Key 变为业务主键，来保障工作解决是有序的。

4. 统计指标依赖明细，服务压力大

另外一个难点就是咱们的很多统计指标都依赖明细，次要是一些实时统计，这种危险比拟显著，服务端压力特地大，尤其是大促时，极其容易把零碎拖垮。

实时超时统计就是一个典型的场景，比如说会有这样两笔订单，一笔订单 1 点钟创立了物流订单，2 点钟进行出库，如何统计超 6 小时未揽收的收单量，因为没有音讯就无奈触发计算，Flink 是基于音讯触发的，比如说 2 点钟出库了，那实践上在 8 点钟的时候超 6 小时未揽收的单量要加 1，然而因为没有音讯触发，上游零碎不会触发计算，这是比拟难的事件，所以一开始没有特地好的计划，咱们间接从明细表出，比方订单的出库工夫是 2 点钟，生成这条明细之后，写到数据库的 OLAP 引擎里，和以后明细进行比拟计算。

咱们也摸索了一些计划比方基于消息中间件，进行一些定时超时音讯下发，或者也摸索过基于 Flink CEP 的形式，第一种形式须要引入第三方的中间件，保护老本会更高，CEP 这种形式采纳工夫窗口稳步向前走，像咱们这种物流场景下会存在很多这样的状况，比方回传一个 2 点出库的工夫，前面发现回传错了，又会补一个 1 点半的工夫，那么咱们须要从新触发计算，Flink CEP 是不能很好的反对的。前面咱们摸索了基于 Flink Timer Service 这种形式，基于 Flink 自带的 Timer Service 回调办法，来制作一个音讯流，首先在咱们的办法外面接入数据流，依据咱们定义的一些规定，比方出库工夫是 2 点，会定义 6 小时的一个超时工夫，注册到 Timer Service 外面，到 8 点会触发一次比拟计算，没有的话就会触发一个超时音讯，整个计划不依赖第三方组件，开发成本比拟低。

5. 履行环节多，数据链路长

另外一个难点就是咱们的履行环节比拟多，数据链路比拟长，导致异常情况很难解决。比方音讯要保留 20 多天的有效期，State 也要存 20 多天，状态始终存在 Flink 外面，如果某一天数据呈现谬误或者逻辑加工谬误，追溯是个很大问题，因为上游的音讯零碎个别放弃三天数据的有效期。

这边说几个实在的案例。

案例 1：

咱们在双十一期间发现了一个 Bug，双十一曾经过来好几天了，因为咱们的履行链路特地长，要 10~20 天，第一工夫发现错误要改曾经改不了了，改了之后 DAG 执行图会发生变化，状态就无奈复原，而且上游只能追 3 天的数，改了之后相当于上游的数全没了，这是不能承受的。

案例 2：

疫情期间的一些超长尾单，State 的 TTL 设置都是 60 天，咱们认为 60 天左右必定可能全副完结，起初发现超过 24 天数据开始失真，明明设置的有效期是 60 天，起初发现底层 State 存储用的是 int 型，所以最多只能存 20 多天的有效期，相当于触发了 Flink 的一个边界 case，所以也证实了咱们这边的场景确实很简单，很多状态须要超长的 State 生命周期来保障的。

案例 3：

每次代码进行降级之后，状态就失落了，须要从新拉取数据计算，然而个别上游的数据只保留 3 天有效期，这样的话业务只能看 3 天的数据，用户体验很不好。

解决方案：批流混合

咱们怎么做？

用批流混合的形式来实现状态复用，基于 Blink 流解决来解决实时音讯流，基于 Blink 的批处理实现离线计算，通过两者的交融，在同一个工作里实现历史所有数据的计算，举个例子，订单音讯流和履行音讯流进行一个关联计算，那么会在工作里减少一个离线订单音讯源，跟咱们的实时订单音讯源 Union All 合并在一起，上面再减少一个 Group By 节点，依照主键进行去重，基于这种形式就能够实现状态复用。有几个须要留神的点，第一个须要自定义 Source Connector 去开发，另外一个波及到离线音讯和实时音讯合并的一个问题，GroupBy 之后是优先取离线音讯还是实时音讯，实时音讯可能生产的比较慢，哪个音讯是真实有效的须要判断一下，所以咱们也定制了一些，比方 LastValue 来解决工作是优先取离线音讯还是实时音讯，整个过程是基于 Blink 和 MaxCompute 来实现的。

1. 一些小的 Tips

① 音讯下发无奈撤回问题

第一个就是音讯一旦下发无奈撤回，所以有些订单一开始无效，前面变成有效了，这种订单不应该在工作中过滤，而是打上标记下传，统计的时候再用。

② 减少数据版本，数据处理工夫以及数据处理版本

• 数据版本是音讯构造体的版本定义，防止模型降级后，工作重启读到脏数据。
• 解决工夫就是音讯以后的解决工夫，比方音讯回流到离线，咱们会依照主键进行工夫排序，取到最新记录，通过这种形式还原一份准实时数据。
• 减少数据处理版本是因为即便到毫秒级也不够准确，无奈辨别音讯的前后程序。

③ 实时对数计划

实时对数计划有两个层面，实时明细和离线明细，刚刚也提到将实时数据回流到离线，咱们能够看以后 24 点前产生的音讯，因为离线 T + 1 只能看到昨天 23 点 59 分 59 秒的数据，实时也能够模仿，咱们只截取那个时刻的数据还原进去，而后实时和离线进行比照，这样也能够很好的进行数据比对，另外能够进行实时明细和实时汇总比照，因为都在同一个 DB 里，比照起来也特地不便。

03 总结与瞻望

1. 总结

简略做下总结：

• 模型与架构：好的模型和架构相当于胜利了 80%。
• 准确性要求评估：须要评估数据准确性要求，是否真的须要对齐 CheckPoint 或者一致性的语义保障，有些状况下保障个别准确性就 ok 了，那么就不须要这么多额定耗费资源的设计。
• 正当利用 Flink 个性：须要正当利用 Fink 的一些个性，防止一些误用之痛，比方 State 和 CheckPoint 的应用。
• 代码自查：保障数据处理是失常流转的，合乎指标。
• SQL 了解：写 SQL 并不是有多高大上，更多考验的是在数据流转过程中的一些思考。

2. 瞻望

① 实时数据品质监控

实时处理不像批处理，批处理跑完之后能够在跑个小脚本统计一下主键是否惟一，记录数稳定等，实时的数据监控是比拟麻烦的事件·。

② 流批对立

流批对立有几个层面，第一个就是存储层面的对立，实时和离线写到同一个中央去，利用的时候更不便。第二个就是计算引擎的对立，比方像 Flink 能够同时反对批处理和流解决，还可能写到 Hive 外面。更高层次的就是能够做到处理结果的对立，同一段代码，在批和流的语义可能会不一样，如何做到同一段代码，批和流的处理结果是齐全对立的。

③ 主动调优

主动调优有两种，比方在大促的时候，咱们申请了 1000 个 Core 的资源，1000 个 Core 怎么正当的调配，哪些地方可能是性能瓶颈，要多调配一些，这是给定资源的主动调优。还有一种比方像凌晨没什么单量，也没什么数据流量，这个时候能够把资源调到很小，依据数据流量状况主动调整，也就是主动伸缩能力。

以上是咱们整体对将来的瞻望和钻研方向。

作者： 张庭（菜鸟数据工程师）

原文链接

本文为阿里云原创内容，未经容许不得转载