通过本文你能 get 到以下知识:
-
Flink 常见的一些维表关联的案例
-
常见的维表方案及每种方案适用场景,优缺点
-
案例:broadcast 实现维表或配置的实时更新
一、案例分析
维表服务在 Flink 中是一个经常遇到的业务场景,例如:
-
客户端上报的用户行为日志只包含了城市 Id,可是下游处理数据需要城市名字
-
商品的交易日志中只有商品 Id,下游分析数据需要用到商品所属的类目
-
物联网温度报警的场景中,处理的是设备上报的一条条温度信息,之前的报警规则是:只要温度大于 20 度需要报警,现在需要改成大于 18 度则报警。这里的报警阈值需要动态调整,因此不建议将代码写死
对于上述的场景,实际上都可以通过维表服务的方式来解决。
二、 维表方案
Flink 中常见的维表方案有以下几种:
1. 预加载维表
在算子的 open 方法中读取 MySQL 或其他存储介质,获取全量维表信息。将维表信息全量保存在内存中。处理数据流时,与内存中的维度进行进行匹配。
例如维度信息保存的是商品 Id 与商品类目的映射关系,那么可以从商品的交易日志中读取出相应的商品 Id,然后去维度表中找出对应的商品类目,将交易日志与商品类目组合起来一块发送给下游。
如果新上架了一些商品 Id 或者某些商品的类目变了,我们无法更新内存里的维度信息。但我们可以在 open 方法中开启一个一分钟一次的定时调度器,每分钟将维度信息读取一次到内存中,从而实现了维度信息的变更。
该方案实现简单,但是有两个很直观的缺陷:
-
维度信息延迟变更:MySQL 中的维度信息随便可能在变,但是只有每分钟才会同步一次
-
维度信息全量加载到内存中:所以不适合维度信息较大的场景。
根据缺陷得出:该方案适用于维表数据量较小,且维表变更频率较低的场景。
当然在 open 方法中我们不只是可以从 MySQL 中去读取,可以自定义各种数据源、各种 DB,甚至可以读取文件,也可以读取 Flink 的 Distributed Cache。
2. 热存储关联
当维度数据较大时,不能全量加载到内存中,可以实时去查询外部存储,例如 MySQL、HBase 等。这样就解决了维度信息不能全量放到内存中的问题,但是对于吞吐量较高的场景,可能与 MySQL 交互就变成了 Flink 任务的瓶颈。每来一条数据都需要进行一次同步 IO,于是优化点就来了:
-
同步 IO 优化为异步 IO
-
对于频繁查找的热数据,可以缓存在内存中,不用每次去查询 MySQL。强烈建议使用 guava 的 Cache 来做缓存。
该方案的优劣势:支持大维度数据量,由于增加了 Cache,可能会导致维度数据更新不及时。
优雅的使用 Cache
Cache 可以认为是功能很丰富的 Map,一般需要设置过期时间,假设 Cache 中设置的 1 s 过期,当缓存中数据存在时,直接查缓存,不查 MySQL,但是在这 1s 内外部 MySQL 中的维度信息可能已经被实时修改了。所以,一定要根据业务场景给 Cache 设定合理的过期时间。对于准确性要求较高的场景过期时间可能要设置在 200ms 以内。
guava 的 Cache 支持两种过期策略,一种是按照访问时间过期,一种是按照写入时间过期。
按照访问时间过期指的是:每次访问都会延长一下过期时间,假如设置的 expireAfterAccess(300, TimeUnit.MILLISECONDS)
,即 300ms 不访问则 Cache 中的数据就会过期。每次访问,它的过期时间就会延长至 300ms 以后。如果每 200ms 访问一次,那么 这条数据将永远不会过期了 。所以一定要注意避坑,如果发现 Cache 中数据一直是旧数据,不会变成最新的数据,可以看看是不是这个原因。
按照写入时间过期指的是:每次写入或者修改都会延迟一下过期时间,可以设置 expireAfterWrite(300, TimeUnit.MILLISECONDS)
表示 300ms 不写入或者不修改这个 key 对应的 value,那么这一对 kv 数据就会被删除。就算在 300ms 访问了 1 万次 Cache,300ms 过期这条数据也会被清理,这样才能保证数据被更新。
对于维度数据不会发生变化的业务场景,按照访问时间过期是最佳的选择。
定义一个 Cache 的代码如下所示:
Cache<Long, Long> cache = CacheBuilder.newBuilder() .maximumSize(1000) // 表示按照访问时间过期 .expireAfterAccess(300, TimeUnit.MILLISECONDS) // 表示按照写入时间过期 .expireAfterWrite(300, TimeUnit.MILLISECONDS) .build();
guava Cache 的功能很丰富,大家可以深入研究其功能及其实现原理。想学习其实现原理的同学,笔者建议先学习 Java 中 LinkedHashMap 的原理。
前面两种方案都存在一个问题:当维度数据发生变化时,更新的数据不能及时更新到 Flink 的内存中,导致线上业务关联到的维表数据是旧数据。那有没有能及时把维度信息通知给 Flink 应用的机制呢?继续往下看,今天的重点来啦。
3. 广播维表
利用 broadcast State 将维度数据流广播到下游所有 task 中。这个 broadcast 的流可以与我们的事件流进行 connect,然后在后续的 process 算子中进行关联操作即可。
当维度信息修改后,我们不只是要把维度信息更新到 MySQL 中,还需要将维度信息更新到 MQ 中。Flink 的 broadcast 流实时消费 MQ 中数据,就可以实时读取到维表的更新,然后配置就会在 Flink 任务生效,通过这种方法及时的修改了维度信息。broadcast 可以动态实时更新配置,然后影响另一个数据流的处理逻辑。
注:广播变量存在于每个节点的内存中,所以数据集不能太大,因为广播出去的数据,会一直在内存中存在。
理论可能理解了,通过案例来深入使用一波。
三、 broadcast 实时更新维表案例
实时处理订单信息,但是订单信息中没有商品的名称,只有商品的 id,需要将订单信息与对应的商品名称进行拼接,一起发送到下游。怎么实现呢?
两个 topic:
-
order_topic_name topic 中存放的订单的交易信息
-
goods_dim_topic_name 中存放商品 id 与 商品名称的映射关系
订单类信息如下所示:
@Data public class Order { /** 订单发生的时间 */ long time; /** 订单 id */ String orderId; /** 用户id */ String userId; /** 商品id */ int goodsId; /** 价格 */ int price; /** 城市 */ int cityId; }
商品信息如下所示:
@Data public class Goods { /** 商品id */ int goodsId; /** 价格 */ String goodsName; }
读取订单交易信息,并从 json 解析为 Order 的过程:
// 读取订单数据,读取的是 json 类型的字符串 FlinkKafkaConsumerBase<String> consumerBigOrder = new FlinkKafkaConsumer011<>("order_topic_name", new SimpleStringSchema(), KafkaConfigUtil.buildConsumerProps(KAFKA_CONSUMER_GROUP_ID)) .setStartFromGroupOffsets(); // 读取订单数据,从 json 解析成 Order 类, SingleOutputStreamOperator<Order> orderStream = env.addSource(consumerBigOrder) // 有状态算子一定要配置 uid .uid("order_topic_name") // 过滤掉 null 数据 .filter(Objects::nonNull) // 将 json 解析为 Order 类 .map(str -> JSON.parseObject(str, Order.class));
读取商品 ID 和 名称的映射信息,从 json 解析成 Goods 类:
// 读取商品 id 与 商品名称的映射关系维表信息 FlinkKafkaConsumerBase<String> consumerSmallOrder = new FlinkKafkaConsumer011<>("goods_dim_topic_name", new SimpleStringSchema(), KafkaConfigUtil.buildConsumerProps(KAFKA_CONSUMER_GROUP_ID)) .setStartFromGroupOffsets(); // 读取商品 ID 和 名称的映射信息,从 json 解析成 Goods 类 SingleOutputStreamOperator<Goods> goodsDimStream = env.addSource(consumerSmallOrder) .uid("goods_dim_topic_name") .filter(Objects::nonNull) .map(str -> JSON.parseObject(str, Goods.class));
定义存储 维度信息的 MapState,将订单流与商品映射信息的广播流进行 connect,进行在 process 中进行关联。process 中,广告流的处理逻辑是:将映射关系加入到状态中。事件流的处理逻辑是:从状态中获取当前商品 Id 对应的商品名称,拼接在一块发送到下游。最后打印输出。
// 存储 维度信息的 MapState final MapStateDescriptor<Integer, String> GOODS_STATE = new MapStateDescriptor<>( "GOODS_STATE", BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO); SingleOutputStreamOperator<Tuple2<Order, String>> resStream = orderStream // 订单流与 维度信息的广播流进行 connect .connect(goodsDimStream.broadcast(GOODS_STATE)) .process(new BroadcastProcessFunction<Order, Goods, Tuple2<Order, String>>() { // 处理 订单信息,将订单信息与对应的商品名称进行拼接,一起发送到下游。 @Override public void processElement(Order order, ReadOnlyContext ctx, Collector<Tuple2<Order, String>> out) throws Exception { ReadOnlyBroadcastState<Integer, String> broadcastState = ctx.getBroadcastState(GOODS_STATE); // 从状态中获取 商品名称,拼接后发送到下游 String goodsName = broadcastState.get(order.getGoodsId()); out.collect(Tuple2.of(order, goodsName)); } // 更新商品的维表信息到状态中 @Override public void processBroadcastElement(Goods goods, Context ctx, Collector<Tuple2<Order, String>> out) throws Exception { BroadcastState<Integer, String> broadcastState = ctx.getBroadcastState(GOODS_STATE); // 商品上架,应该添加到状态中,用于关联商品信息 broadcastState.put(goods.getGoodsId(), goods.getGoodsName()); } }); // 结果进行打印,生产环境应该是输出到外部存储 resStream.print();
通过上述代码,已经完成了我们的需求,生产环境中将结果输出到外部存储即可。
小优化点:(小但是非常有必要的优化)
商品下架,就不会再有该商品的交易信息,此时应该将商品从状态中移除,防止状态无限制的增大。怎么设计呢?
首先对 Goods 类进行重新定义,增加了 isRemove 字段,要来标识当前商品是上架还是下架,如果下架应该从 State 中去移除:
@Data public class Goods { /** 商品id */ int goodsId; /** 价格 */ String goodsName; /** * 当前商品是否被下架,如果下架应该从 State 中去移除 * true 表示下架 * false 表示上架 */ boolean isRemove; }
BroadcastProcessFunction 的 processBroadcastElement 也应该改动,判断如果是上架,应该添加到状态中,用于关联商品信息。如果商品下架,应该要从状态中移除,否则状态将无限增大。
// 更新商品的维表信息到状态中 @Override public void processBroadcastElement(Goods goods, Context ctx, Collector<Tuple2<Order, String>> out) throws Exception { BroadcastState<Integer, String> broadcastState = ctx.getBroadcastState(GOODS_STATE); if (goods.isRemove()) { // 商品下架了,应该要从状态中移除,否则状态将无限增大 broadcastState.remove(goods.getGoodsId()); } else { // 商品上架,应该添加到状态中,用于关联商品信息 broadcastState.put(goods.getGoodsId(), goods.getGoodsName()); } }
当维度信息较大,每台机器上都存储全量维度信息导致内存压力过大时,可以考虑进行 keyBy,这样每台节点只会存储当前 key 对应的维度信息,但是使用 keyBy 会导致所有数据都会进行 shuffle。当然上述代码需要将维度数据广播到所有实例,也是一种 shuffle,但是维度变更一般只是少量数据,成本较低,可以接受。大家在开发 Flink 任务时应该根据实际的业务场景选择最合适的方案。
四、 总结
开篇介绍了一些需要使用维表的场景,然后讲述了常见的维表方案及每种方案适用场景,优缺点。最后着重通过一个案例给大家详细介绍了如何使用 broadcast 实现维表或配置的实时更新,并给出了一些优化点。维表关联属于面试常见考点,且 broadcast 实现维表关联非常受面试官的欢迎,希望本文对大家有所帮助。
本文详细代码请参考,感觉有用的同学帮忙点一波 Star:https://github.com/1996fanrui/fanrui-learning/blob/master/module-flink/src/main/java/com/dream/flink/connect/BroadcastOrderJoinGoodsName.java
同时再推荐大家阅读我的上一篇实战文章: Flink 使用 connect 实现双流匹配 以及我公众号底部的 Flink 系列文章,希望对大家有所帮助。
近期还会更新实战案例,都是面试高频考点哦:
-
Flink 实时计算分组 TopN 的问题,例如求每个城市最活跃的那 100 个用户
以上就是本文的全部内容,希望本文的内容对大家的学习或者工作能带来一定的帮助,也希望大家多多支持 码农网
猜你喜欢:- probie 插件更新,新增实时编程
- DuiC 配置中心 1.7.0 发布,支持配置实时更新
- 如何增强推荐系统模型更新的实时性?
- 亿级视频内容如何实时更新?(附阿里文娱智能算法PDF下载)
- vue实现歌手列表字母排序,下拉滚动条侧栏排序实时更新
- 配置中心 duic-java-client 1.3.0 发布,配置实时更新
本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们。