流计算技术实战超大维表问题

维度表&＃xff0c;作为数据仓库里面的概念&＃xff0c;是维度属性的集合&＃xff0c;比如时间维、地点维;

但这里要讨论流计算中的维度表问题&＃xff0c;

流计算中维表问题和数据仓库中有所不同&＃xff0c;往往是因为通过agent采集到的数据比较有限&＃xff0c;在做数据业务的时候&＃xff0c;需要先实时的把这些维度信息给补全&＃xff1b;

这个问题其实就是&＃xff0c;主数据流和多个静态表或半静态表之间的join问题。

在flink中称为side input问题&＃xff0c;https://cwiki.apache.org/confluence/display/FLINK/FLIP-17&＃43;Side&＃43;Inputs&＃43;for&＃43;DataStream&＃43;API

解决维表问题考虑到点&＃xff0c;

a. 对元数据库的读压力&＃xff1b;如果分析程序有1000并发&＃xff0c;是否需要读1000次

b. 读维表数据不能拖慢主数据流的throughput&＃xff0c;每秒千万条数据量

c. 动态维表更新问题和一致性问题&＃xff1b;元数据是不断变化的&＃xff0c;如何把更新同步到各个并发上

d. 冷启动问题&＃xff0c;如何保证主数据流流过的时候&＃xff0c;维表数据已经ready&＃xff0c;否则会出现数据无法处理

e. 超大维表数据会导致流量抖动和频繁gc&＃xff0c;比如几十万条的实例数据&＃xff0c;可能上百兆

下面谈谈我们解决这个问题的思路&＃xff0c;

1. 最简单的版本&＃xff0c;每个进程都会独立的去从元数据里面读取元数据&＃xff1b;

这样的优点是简单&＃xff0c;c&＃xff0c;d问题天然解决&＃xff1b;但只能适用于数据量较小的场景&＃xff0c;否则并发太大&＃xff0c;a&＃xff0c;肯定就无法满足

2. 随着业务量的扩大&＃xff0c;处理程序的并发越来越大&＃xff0c;1&＃xff0c;很快会达到瓶颈

我们就采用新的方案&＃xff0c;这个方案是在Jstorm环境实现的&＃xff0c;用一个spout读&＃xff0c;然后广播给所有的处理进程

这个方案主要解决a&＃xff0c;c的问题&＃xff0c;
但是也引入了d&＃xff0c;e的问题&＃xff0c;

解决d&＃xff0c;Jstorm支持让某个spout在job启动后等待一段时间&＃xff0c;所以可以让主数据流spout等待几分钟再开始读数据&＃xff0c;这样保证数据到的时候&＃xff0c;维表数据已经ready&＃xff1b;这个解法每次重启job都要等好几分钟&＃xff0c;体验挺差的&＃xff0c;但是勉强可以work

e问题&＃xff0c;一个spout广播超大维表到几百并发的线程&＃xff0c;首先就是会队列满&＃xff0c;因为jstorm发一份数据到所有并发的时候&＃xff0c;是需要产生几百份真实数据在队列中的&＃xff1b;然后GC也会很严重&＃xff0c;因为大量的临时对象会产生释放&＃xff0c;在传输和进程cache过程中&＃xff0c;会导致业务抖动

这个问题只能增加内存和worker数来解决&＃xff0c;否则job有可能会完全hang死

我们也用Chronicle Map(https://github.com/OpenHFT/Chronicle-Map)来尝试解决内存使用和gc的问题

BTW&＃xff0c;有同学问&＃xff0c;如果让数据和维表数据都 shuffle by key&＃xff0c;是不是可以缓解这个问题
如果数据量比较小&＃xff0c;可以考虑&＃xff0c;但是对于我们的主数据流的数据量&＃xff0c;是没法shuffle的&＃xff0c;所以需要在每个并发上保留全量的维表信息

2.1 用Flink带替换Jstorm

Flink虽然在window&＃xff0c;乱序&＃xff0c;一致性等方面做了很大的改进&＃xff0c;但是在这个问题上仍然没有很好的解&＃xff0c;上面提到的side input也没有实现出来&＃xff1b;

并且Flink随着更多的高层的封装&＃xff0c;程序员的开发自由度是降低的&＃xff0c;和JStorm比&＃xff0c;所以如果用Flink解决上面的问题&＃xff0c;没有本质变化&＃xff0c;可能JStorm更麻烦&＃xff1b;

需要用ConnectedStreams去joine数据流和side input流&＃xff0c;
对于d问题&＃xff0c;没法直接解决
对于e问题&＃xff0c;因为flink对内存管理做的比较好&＃xff0c;gc问题有所缓解&＃xff0c;但是job抖动的问题还是会存在

因为广播这么大的数据&＃xff0c;会中断主数据流的数据处理&＃xff0c;也会大大增加checkpoint的时间&＃xff0c;如下图&＃xff0c;可以看到30分钟一次的同步

BTW&＃xff0c;Flink保障一致性&＃xff0c;提供checkpoint机制&＃xff0c;但也增加复杂性&＃xff0c;这个地方处理不好会有很多问题
比如&＃xff0c;如果在source中同步读数据库数据&＃xff0c;如果读库的时间比较长&＃xff0c;就会hang住主数据流&＃xff0c;因为其他operator都会等它完成checkpoint&＃xff0c;写JStorm的程序员需要注意这点&＃xff0c;Flink需要更精细的控制&＃xff0c;任何operator&＃xff0c;任何并发的hang都会导致整个任务hang

我个人尝试使用flink本身的机制&＃xff0c;statebackend&＃xff0c;rocksdb等来更优雅的解决这个问题&＃xff0c;但是没有发现比较好的方法&＃xff0c;或者实现过于复杂

3. Redis版本

这其实是把1&＃xff0c;2方法做了综合

使用redis来做cache&＃xff0c;只用一个job&＃xff0c;负责从元数据库同步数据到redis&＃xff0c;这样就解决a&＃xff0c;c

然后所有的并发都从redis直接查询需要的元数据&＃xff0c;这样就解决d

对于b&＃xff0c;在并发上做local cache&＃xff0c;只有第一次需要真正查询redis&＃xff0c;后续定期异步更新就好&＃xff0c;不会影响到主数据流

对于e&＃xff0c;因为现在不需要一下全量的读取维表数据到内存&＃xff0c;用到的时候才去读&＃xff0c;分摊了负载&＃xff0c;也可以得到缓解

该方案当前线上跑着&＃xff0c;还算比较稳定

这个方案最大的缺点是增加依赖&＃xff0c;对于需要全球多region大规模部署的应用&＃xff0c;增加依赖是成本极高的
同时要额外保障redis和同步job的稳定性

BTW&＃xff0c;这里不建议local cache用LRU&＃xff0c;因为要考虑到当redis挂掉或同步job挂掉的时候&＃xff0c;不能影响主数据流&＃xff0c;所以我只会异步的更新local cache&＃xff0c;但不去做过期&＃xff0c;这样就算redis挂了&＃xff0c;也只是影响更新的实例&＃xff0c;大大降低故障发生概率

总结&＃xff0c;

分享一下自己的一些实战经验&＃xff0c;希望可以抛砖引玉&＃xff0c;找到更合理&＃xff0c;优雅的方案