FlinkSQL在京东实时维度建模中的应用

导读&＃xff1a;今天给大家分享京东在实时建模过程中使用Flink SQL的应用案例。主要从问题、难点、方案和规划四个方面来介绍。

01

问题

在维度建模中有很多技术场景&＃xff0c;本文挑选其中两个较难的场景来进行详细阐述。

问题一 实时多流全量关联的问题

select * from A full join B on A.name &＃61; B.name;

此问题关联SQL比较简单&＃xff0c;只有A流和B流两个实时流进行full join&＃xff0c;但还可能涉及到其中有一流数据需要关联全量历史数据。

问题二 实时流全量分组计算的问题

select id,name,val,row_number() over(partition byname order by val) as rn from A;
select name,min(val) from A group by k;

这个问题是实时流全量分组计算&＃xff0c;涉及到分组排序值&＃xff0c;或者分组求最小值等操作。

针对以上两个问题&＃xff0c;似乎使用SQL处理起来比较简单&＃xff0c;甚至我们可以直接用Flink SQL去完成。而实际直接使用以上Flink SQL计算存在若干问题。第一个问题是在最初生产数据的时候&＃xff0c;可能并没有接入实时数据流。在实际场景中&＃xff0c;往往需要在过程中接入实时数据。比如接入binlog之类的数据&＃xff0c;其实是从当前时刻到未来这一时间段。例如商品SPU信息&＃xff0c;一般需要有历史数据。只有获取历史数据&＃xff0c;在订单数据流关联的时候&＃xff0c;才能关联上这个订单所对应的历史信息。如果商品信息都是从产生后一段时间才接入数据&＃xff0c;即从当前时刻才有的&＃xff0c;那就无法获取到历史信息。第二个问题是状态数据&＃xff0c;在用Flink SQL进行计算时&＃xff0c;需要维护若干状态数据放到内存中&＃xff0c;这种操作比较消耗内存。对于商品几年的历史数据&＃xff0c;如果将状态数据存放在rocksdb中&＃xff0c;经测试性能无法得到保障。基于此&＃xff0c;如果直接用Flink SQL肯定无法解决。

以下分析怎么去解决此类问题。

首先看第一个问题&＃xff0c;即实时多流全量关联的问题。

第一部分就是从最初时刻到当前时刻的时间区间&＃xff0c;可以认为是历史状态数据。此处分别是A表和B表的记录。然后我们按照对应的关联键&＃xff0c;如分别按name进行关联。通过name字段把它拆成几个部分&＃xff0c;右侧是关联出来的结果表。当有一个消息进入&＃xff0c;这个消息体可能是一个变更的消息记录。此处略过较为简单的新增或者删除操作&＃xff0c;重点分析更新操作。

比如下图中消息mid&＃61;1&＃xff0c;将A表里边的id&＃61;1000001这条记录里面aaa的记录值改成了bbb。但这个name字段是关联键&＃xff0c;所以这就导致它影响了两个分组下的关联键下的记录。最终在A里面&＃xff0c;它与B关联上的&＃xff0c;之前的这两条记录(name&＃61;aaa)&＃xff0c;在它被修改之后就不存在了&＃xff0c;所以我们需要给下游发删除记录。

同时在A表的分组中&＃xff0c;即在name&＃61;bbb分组下&＃xff0c;这个关联键下会新增一条插入的记录。所以需要向下游下发这条新增记录跟B表关联到的这两条宽表记录。如果再有一条消息体过来&＃xff0c;那么同样存在插入和删除记录。以此类推&＃xff0c;B表变更逻辑相类似。在此过程中&＃xff0c;当一条变更的消息体来了之后&＃xff0c;需要考虑在实时维度建模里面去做哪些操作。例如在这个建模过程中需要变更的可能不只是当前的那个记录状态&＃xff0c;之前的状态也有关联。而在进行分布式去计算的时候&＃xff0c;不同的关联键也会分发到不同的节点上去计算。

再看第二个问题&＃xff0c;全量分组计算问题。

类似地&＃xff0c;当源端来了一个消息体&＃xff0c;本来是一个原始状态值name&＃61;bbb&＃xff0c;当它变更之后&＃xff0c;现在变更成name&＃61;aaa&＃xff0c;最终会影响分组排序的结果。原本是按照val&＃61;21,33,66 (name&＃61;aaa)依次排序&＃xff0c;现在只有val&＃61;33,66 (name&＃61;bbb)&＃xff0c;因为&＃xff08;id&＃61;100003&＃xff09;这条记录中的name值从bbb变成了aaa。所以在排序时row_number会发生变更&＃xff0c;而下发的相关联的数据同样都会变更。当产生变更时&＃xff0c;分组中最小值(min)是一样的&＃xff0c;在涉及变更行为之后都需要更改。我们在进行计算时就需要考虑以上这些情况。因此每一次计算过程&＃xff0c;都需要拿到A表的全量历史数据跟B表的全量历史数据去进行关联计算。

当然也可能存在A表发生变更&＃xff0c;需要拿关联的B表的全量&＃xff0c;这样可以把所有的数据都进行关联&＃xff0c;包括之前这个状态将对应的B的全量取出来然后去做关联。以上这两类问题&＃xff0c;通过画图和描述来看&＃xff0c;似乎较为简单。但问题在于有数据流的变更&＃xff0c;还需要还原出对应的状态表&＃xff0c;然后由状态表关联结果&＃xff0c;最终还原出对应的关联结果的流式变更记录&＃xff0c;这类操作就涉及状态存储与读取的技术难点。

02

难点

以上两类问题&＃xff0c;如果进行全量数据关联操作&＃xff0c;很快就能得到关联出的所有全量结果&＃xff0c;然后去做计算&＃xff0c;在性能上似乎也不用考虑太多&＃xff0c;但实际操作存在一定难度。以下阐述了问题难点。

首先需要明确&＃xff0c;问题难点准确来说是相对繁琐之处。实时维度建模有别于离线维度建模&＃xff0c;需要考虑的地方很多&＃xff0c;因此会增加实时维度建模的开发难度以及需要人为的干涉的地方。

难点一&＃xff0c;如何获取到历史数据。

如果是使用Flink SQL来实现&＃xff0c;需要保存全量的历史信息。既然我们无法保存状态&＃xff0c;那就存到外设KV存储里面&＃xff0c;存到外部的KV Storage。在这个过程中&＃xff0c;需要考虑以什么结构去存储&＃xff1b;如何去维护A、B、C三个表的数据&＃xff0c;用这三个流表去做关联&＃xff1b;这三个流表与各自对应的外设存储比如说A的HBase表跟B的HBase表跟C的HBase表或者与D的HBase表等等类似KV存储这种类型的&＃xff1b;要对这几个表之间的关系进行组织&＃xff1b;在组织关联出了结果之后&＃xff0c;需要考虑在关联结果里面是否还存在时序问题。

如上描述的关联结果里面存在三个消息&＃xff0c;当三个消息体来了之后&＃xff0c;对应的有一组下发流。如id&＃61;100001对应的键&＃xff0c;它存在多条记录&＃xff0c;如果时序发生混乱&＃xff0c;最终可能导致宽表的结果还原错误。那怎么来保证时序的正确性&＃xff0c;如果直接用Flink SQL进行计算&＃xff0c;可能需要创建一个HBase table source&＃xff08;类似table function&＃xff09;的方式去做关联。

难点二&＃xff0c;如何提升性能。

如果直接使用Flink SQL进行计算&＃xff0c;需要考虑如何获取数据包括IO操作&＃xff0c;且如何更高效地去实现这个过程。关于提升性能&＃xff0c;最直接地有两种途径&＃xff0c;一是降低IO的请求次数&＃xff0c;二是增加并发。

降低IO请求&＃xff0c;具体实现方法是做业务逻辑过滤或者根据业务逻辑做一定计算&＃xff0c;或者做一些请求合并&＃xff0c;或者用一些批量请求的方式去发请求。另外是设置缓存&＃xff0c;通过设置一个合理的缓存值&＃xff0c;能够做到缓存有一个较高的命中率。同时还需要考虑是设置被动缓存还是主动缓存。

增加并发&＃xff0c;具体需要考虑是使用异步IO&＃xff0c;还是使用增加并行度、引擎优化等等。因为以上两个较为繁琐支出&＃xff0c;所以开发难度相对较高&＃xff0c;需要考虑的细节问题较多。如果直接使用Flink SQL&＃xff0c;其代价也许比写Flink程序去做维度建模工作更大。以上就是使用Flink SQL做维度建模工作遇到的若干难题。

03

方案

那如何去解决以上问题&＃xff1f;首先考虑采用组件化的设计方式&＃xff0c;把用户所需要的维度建模功能做成若干组件&＃xff0c;用户可以直接通过组件化的配置&＃xff0c;就能够完成维度建模的开发。这样达到了降低代码开发成本的目的。

其次是将计算逻辑使用Flink SQL来表达。因为SQL在数据开发过程中是一种通用性较强的方式&＃xff0c;并且其学习成本或研发、维护等等成本相对较低。采用以上两种方式需要解决前文提到的难点&＃xff0c;下面看具体方案。

这里构建了一个产品RDM Building用来完成维度建模构建工作。它主要分为三层&＃xff08;Components Config、RDDM Component Builder、RDDM Component Parse&＃xff09;&＃xff0c;首先用户可以写组件配置&＃xff0c;然后构建器会根据配置生成对应的一些组件对象&＃xff0c;接着组件对象通过RDDM Component Parse就会将对应的组件转化Flink SQL算子。

例如使用A、B、C三个流进行关联计算&＃xff0c;它们都是通过同样一组key关联&＃xff0c;其关联键是相同的&＃xff0c;最终关联生成一个结果表。这类操作可以使用His Rows Component组件进行处理。

它负责的过程是将A、B、C流分别对应的KV存储里面存储的历史数据取出&＃xff0c;拿到对应的关联历史数据&＃xff0c;比如A流数据过来就取到对应的关联的B表跟C表的历史数据。同样的针对B流&＃xff0c;会将关联的A流跟C流的历史信息也都取出。对于C表也是如此。这样需要关联的所有历史数据&＃xff0c;再组合成对应的A&＃39;、 B&＃39;、C&＃39;&＃xff0c;其种A&＃39;不仅包含了A当前的数据&＃xff0c;还包含了A要跟其它流关联的、相关的历史数据。所以在A&＃39;、 B&＃39;、C&＃39;里面&＃xff0c;不仅有A、B、C流数据&＃xff0c;而且包含了对应的历史数据。

按照以上设计&＃xff0c;实时流的历史数据可以一直保存&＃xff0c;那就基本解决历史数据这个难点。还需要考虑的问题是cache随着时间增长存不下&＃xff0c;或者因cache太大导致速度很慢。又或者使用rocksdb的方式发现性能较低&＃xff0c;但同时缺乏rocksdb方面优化的专业能力。但可以相对较为成熟的KV存储&＃xff0c;而且数据可能本身使用KV存储进行保存。那么直接使用KV存储就能解决历史数据问题。

下图可以看到His Rows Component组件的详细执行过程&＃xff0c;我们以这个组件为例来进行展开分析。

首先&＃xff0c;我们拿到A、B、C三个流表&＃xff0c;拿到之后先根据其对应的组件去做keyBy操作&＃xff0c;再去做窗口聚合。窗口聚合的功能是什么&＃xff1f;在实际业务过程中&＃xff0c;A表的业务数据在处理过程中可能存在事务操作。它的事务操作里面可能存在先签了一条记录&＃xff0c;然后再把里面改了一下某一个值&＃xff1b;也可能先删了一条记录&＃xff0c;然后又插入了一条记录。这取决于业务系统的多样性操作&＃xff0c;所以可能存在业务上只做了一次变动&＃xff0c;但发了若干条消息对应的binlog。那就需要首先对其进行微批处理&＃xff0c;来减少我们处理的记录数&＃xff0c;然后处理生成A、B、C一一对应的三个流。

接着再把它们进行格式转换&＃xff0c;转换成对应的tuple。比如相同的一个格式&＃xff0c;可能是表名、分组键、主键以及转换成的统一的Bean结构。此处需要注意的问题是&＃xff0c;对更新这种操作需要拆分。具体有两类&＃xff0c;一类是插入的tuple&＃xff0c;一类是删除的tuple&＃xff0c;它会有对应两条记录&＃xff0c;一条删除的一条新增的。在统一格式之后&＃xff0c;再把它们union(合并)到同一个流&＃xff0c;然后根据相同的关联键去做keyBy。这个过程中合流统一keyBy&＃xff0c;是想用这种方式把所有的需要关联的流数据通过相同的键去做keyBy然后做分发&＃xff0c;保证的结果是使相同存储键的记录分发到相同slot&＃xff0c;以提升缓存利用率。从KV Storage获取数据&＃xff0c;或者过程中使用的缓存的高命中率&＃xff0c;包括缓存的有效性&＃xff0c;以及缓存一致性&＃xff0c;即最终同样的一个key一定会落到同一个slot上去计算。这样就能够保证不会存在多个slot上都存了同一个关联键的一些信息&＃xff0c;从而可以避免不一致或先后问题。

得到keyBy之后的结果&＃xff0c;下一步经过flatmap算子&＃xff0c;可以把历史数据从cache缓存、KV存储中加载进来。加载进来之后再比较它们的消息的id&＃xff0c;如果是最新记录则就写入cache及KV存储并向下发送记录。当A流来了&＃xff0c;关联的是B流历史跟C流历史&＃xff1b;当B流来了&＃xff0c;同样会关联A流历史和C流历史。同理C流也是如此。

下一步把从hisStreams流中分拆加载出来的数据流&＃xff0c;合并得到包含历史数据的A_all、B_all、C_all 中&＃xff0c;并用于下一步SQL计算。这个功能是将A、B、C流映射成了A_all、B_all、C_all&＃xff0c;即所有的SQL逻辑就都可以直接写Flink SQL并用这三个流表(A_all、B_all、C_all)去计算&＃xff0c;如此就包含短时间内的历史数据在进行计算。当然如果缓存中有数据的话&＃xff0c;就不需要从KV中加载。而且同样的key会发到同一个slot上&＃xff0c;它的缓存的命中率是很高的。另外以上都是按关联键去存的&＃xff0c;如果同时有A、B、C等多条消息来了之后&＃xff0c;都会命中同一条KV存储的key&＃xff0c;也会提升缓存命中率。

04

规划

以上是具体实现的方案&＃xff0c;下面是我们后续的规划。

目前在用户使用以上组件时&＃xff0c;还存在较高的学习成本。后续我们可能会增加一个前端页面&＃xff0c;可以直接生成基于类似流程图面板以及点击、拖拉拽等形式生成对应不同类型的组件。采用这种配置方式可以降低用户的学习成本&＃xff0c;增加使用的便捷性。另外我们还要扩展底层对多种实时计算引擎的支持。现在采用RDDM Component Parse会把对应的组件转成Flink算子(keyBy、flatMap等)&＃xff0c;但可能在某些应用场景下还会使用SparkStreaming&＃xff0c;或其它实时计算引擎。针对这种场景&＃xff0c;我们可以将其独立出来&＃xff0c;供用户自由地选择实时计算引擎。在KV存储方面&＃xff0c;目前以内部使用为主&＃xff0c;暂时只支持HBase、redis这两类KV存储&＃xff0c;也需要把这部分信息彻底地独立。

如果这个文章对你有帮助&＃xff0c;不要忘记 「在看」 「点赞」 「收藏」 三连啊喂&＃xff01;

2022年全网首发|大数据专家级技能模型与学习指南(胜天半子篇)

互联网最坏的时代可能真的来了

我在B站读大学&＃xff0c;大数据专业

我们在学习Flink的时候&＃xff0c;到底在学习什么&＃xff1f;

193篇文章暴揍Flink&＃xff0c;这个合集你需要关注一下

Flink生产环境TOP难题与优化&＃xff0c;阿里巴巴藏经阁YYDS

Flink CDC我吃定了耶稣也留不住他&＃xff01;| Flink CDC线上问题小盘点

我们在学习Spark的时候&＃xff0c;到底在学习什么&＃xff1f;

在所有Spark模块中&＃xff0c;我愿称SparkSQL为最强&＃xff01;

硬刚Hive | 4万字基础调优面试小总结

数据治理方法论和实践小百科全书

标签体系下的用户画像建设小指南

4万字长文 | ClickHouse基础&实践&调优全视角解析

【面试&个人成长】2021年过半&＃xff0c;社招和校招的经验之谈

大数据方向另一个十年开启 |《硬刚系列》第一版完结

我写过的关于成长/面试/职场进阶的文章

当我们在学习Hive的时候在学习什么&＃xff1f;「硬刚Hive续集」