flink1.12.x 重分区算子-broadcast() 广播

作者：马育民 • • 阅读：10122

提出问题

有一个 变量 是 大数组，名字是： BigArr ，数据量很大，需要占用 1G 内存。

如果每个节点有 两个task，那么 每个task 就有一个 BigArr，那么 每个节点 就会有 2个 BigArr，占用 2G内存

如下图：

分析

如果程序 只是对 BigArr 进行 读操作，而 不进行修改操作，那么 一个 节点 的 多个 task 可以共用 一个BigArr

否则：占用内存太大了，而且是没有必要的

解决

使用 broadcast() 广播

broadcast() 广播

上游 每一个数据，都会分发给下游 所有分区

一句话理解

公共的共享变量，把数据集广播出去，在不同的 task 上都能够获取到，每个节点上只会存在一份。

注意：不应该修改广播变量，这样才能确保每个节点获取到的值都是一致的。

解决上面问题

应用场景

  • 规则或配置更新：当有新的业务规则或配置需要实时推送到所有计算节点时，可以使用广播流来高效地分发这些变化。

  • 维表关联：在流处理中，有时需要将流数据与维度表（通常是相对静态的大表）进行关联。广播流可以用来广播维度表数据到所有任务，实现类似于数据库join的操作，而无需每个事件都查询外部系统。

  • 事件驱动的应用：在事件驱动架构中，全局事件（如系统状态变更通知）可以通过广播流发送，确保所有相关组件都能及时响应。

案例

数据流，经过下面处理：

  • 执行 map() 算子，传入 RichMapFunction 对象，将 任务索引 和 原数据 组装，然后打印结果。
    通过结果看出：同一个数据只会进入下游的某一个任务中

  • 执行 broadcast() 广播后，在执行 map() 算子，传入 RichMapFunction 对象，将 任务索引 和 原数据 组装，然后打印结果。
    通过结果看出：广播后，同一个数据会进入下游的每一个任务中

生成可视化图

访问：https://flink.apache.org/visualizer/

package chongfenqu;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class F11broadcast {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度是2
        env.setParallelism(2);

        DataStreamSource<String> dss = env.fromElements("lilei", "lili", "lucy", "hanmeimei");

        // 普通算子，与下面广播做对比
        DataStream<Tuple2<Integer,String>> mapDS = dss.map(new RichMapFunction<String, Tuple2<Integer,String>>() {
            @Override
            public Tuple2<Integer, String> map(String s) throws Exception {
                return Tuple2.of(getRuntimeContext().getIndexOfThisSubtask(),s);
            }
        });

        mapDS.print("mapDS::");


        DataStream<Tuple2<Integer,String>> broadcastDS = dss
                .broadcast()// 广播
                .map(new RichMapFunction<String, Tuple2<Integer,String>>() {
            @Override
            public Tuple2<Integer, String> map(String s) throws Exception {
                return Tuple2.of(getRuntimeContext().getIndexOfThisSubtask(),s);
            }
        });

        broadcastDS.print("broadcastDS::");

        System.out.println(env.getExecutionPlan());
        env.execute();
    }
}

执行结果

mapDS:::1> (0,lilei)
broadcastDS:::1> (0,lilei)
mapDS:::1> (0,lucy)
mapDS:::2> (1,lili)
broadcastDS:::2> (1,lilei)
mapDS:::2> (1,hanmeimei)
broadcastDS:::2> (1,lili)
broadcastDS:::1> (0,lili)
broadcastDS:::2> (1,lucy)
broadcastDS:::1> (0,lucy)
broadcastDS:::2> (1,hanmeimei)
broadcastDS:::1> (0,hanmeimei)

可以看到：

普通算子 mapDS:::：

  • 将 数据 lilei、lucy 发送到 0号分区

  • 将 数据 lili、hanmeimei 发送到 1号分区

广播算子 broadcastDS:::：

将 所有数据，发送到 所有分区

参考

https://blog.51cto.com/u_9928699/5670559
https://juejin.cn/post/7386873708600180786

原文出处：/show_1IX44a8HJjwn.html

吉ICP备15003043号