简介：供应链物流场景下的业务复杂度高，业务链路长，节点多，实体多，实时数仓建设难度高。菜鸟跨境进口业务场景更是如此，更简单的场景带来更简单的实体数据模型，对接的业务零碎多导致ETL流程特地简单，还有海量的日均解决数据量，使得团队在建设进口实时数仓的过程中，面临着诸多挑战。
如何保障简单实体关系下的数据准确性？如何升高多数据源状况下的数据处理复杂度？如何晋升实时多流Join的解决效率？如何实现实时超时统计？如何实现异常情况下的数据状态复原？本文次要分享菜鸟进口实时数仓的降级教训，以及如何利用Flink的个性解决在开发实际中遇到的问题。
次要内容包含：
- 相干背景介绍
- 进口实时数仓演进过程
- 挑战及实际
- 总结与瞻望
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和音讯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来实现的。
- 一些小的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怎么正当的调配，哪些地方可能是性能瓶颈，要多调配一些，这是给定资源的主动调优。还有一种比方像凌晨没什么单量，也没什么数据流量，这个时候能够把资源调到很小，依据数据流量状况主动调整，也就是主动伸缩能力。
以上是咱们整体对将来的瞻望和钻研方向。
作者：张庭（菜鸟数据工程师）
原文链接
本文为阿里云原创内容，未经容许不得转载