Flink 使用 broadcast 实现维表或配置的实时更新

问题导读

1.本文介绍了几种维表方案？
2.各个方案有什么优缺点？
3. broadcast如何实现实时更新维表案例？



通过本文你能 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 的代码如下所示：

[mw_shl_code=bash,true]Cache<Long, Long> cache = CacheBuilder.newBuilder()
        .maximumSize(1000)
        // 表示按照访问时间过期
        .expireAfterAccess(300, TimeUnit.MILLISECONDS)
        // 表示按照写入时间过期
        .expireAfterWrite(300, TimeUnit.MILLISECONDS)
        .build();[/mw_shl_code]

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 与 商品名称的映射关系
  

订单类信息如下所示：

[mw_shl_code=bash,true]@Data
publicclass Order {
    /** 订单发生的时间 */
    long time;

    /** 订单 id */
    String orderId;

    /** 用户id */
    String userId;

    /** 商品id */
    int goodsId;

    /** 价格 */
    int price;

    /** 城市 */
    int cityId;
}[/mw_shl_code]

商品信息如下所示：
[mw_shl_code=bash,true]@Data
publicclass Goods {
    /** 商品id */
    int goodsId;

    /** 价格 */
    String goodsName;
}[/mw_shl_code]


读取订单交易信息，并从 json 解析为 Order 的过程：
[mw_shl_code=bash,true]// 读取订单数据，读取的是 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));[/mw_shl_code]

读取商品 ID 和 名称的映射信息，从 json 解析成 Goods 类：
[mw_shl_code=bash,true]// 读取商品 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));[/mw_shl_code]


定义存储 维度信息的 MapState，将订单流与商品映射信息的广播流进行 connect，进行在 process 中进行关联。process 中，广告流的处理逻辑是：将映射关系加入到状态中。事件流的处理逻辑是：从状态中获取当前商品 Id 对应的商品名称，拼接在一块发送到下游。最后打印输出。
[mw_shl_code=bash,true]// 存储 维度信息的 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();[/mw_shl_code]

通过上述代码，已经完成了我们的需求，生产环境中将结果输出到外部存储即可。

小优化点：（小但是非常有必要的优化）
商品下架，就不会再有该商品的交易信息，此时应该将商品从状态中移除，防止状态无限制的增大。怎么设计呢？

首先对 Goods 类进行重新定义，增加了 isRemove 字段，要来标识当前商品是上架还是下架，如果下架应该从 State 中去移除：

[mw_shl_code=bash,true]@Data
publicclass Goods {

    /** 商品id */
    int goodsId;

    /** 价格 */
    String goodsName;

    /**
     * 当前商品是否被下架，如果下架应该从 State 中去移除
     * true 表示下架
     * false 表示上架
     */
    boolean isRemove;
}[/mw_shl_code]
BroadcastProcessFunction 的 processBroadcastElement 也应该改动，判断如果是上架，应该添加到状态中，用于关联商品信息。如果商品下架，应该要从状态中移除，否则状态将无限增大。
[mw_shl_code=bash,true]// 更新商品的维表信息到状态中
@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());
    }
}[/mw_shl_code]


当维度信息较大，每台机器上都存储全量维度信息导致内存压力过大时，可以考虑进行 keyBy，这样每台节点只会存储当前 key 对应的维度信息，但是使用 keyBy 会导致所有数据都会进行 shuffle。当然上述代码需要将维度数据广播到所有实例，也是一种 shuffle，但是维度变更一般只是少量数据，成本较低，可以接受。大家在开发 Flink 任务时应该根据实际的业务场景选择最合适的方案。

四、 总结

开篇介绍了一些需要使用维表的场景，然后讲述了常见的维表方案及每种方案适用场景，优缺点。最后着重通过一个案例给大家详细介绍了如何使用 broadcast 实现维表或配置的实时更新，并给出了一些优化点。维表关联属于面试常见考点，且 broadcast 实现维表关联非常受面试官的欢迎，希望本文对大家有所帮助。

本文详细代码请参考，感觉有用的同学帮忙点一波 Star：

https://github.com/1996fanrui/fa ... rJoinGoodsName.java




原文链接
https://mp.weixin.qq.com/s/IyhKPerbzy-9K8aJq5FQdQ

感谢分享

不错，学习了