在Flink中，同一个算子可能存在若干个不同的并行实例，计算过程可能不在同一个Slot中进行，不同算子之间更是如此，因此不同算子的计算数据之间不能像Java数组之间一样互相访问，而广播变量Broadcast便是解决这种情况的.
在 flink 中, 针对某一个算子需要使用公共变量的情况下, 就可以把对应的数据给广播出去, 这样在所有的节点中都可以使用了. 典型的代码结构如下所示:
在一个算子中使用广播变量主要有两个步骤:
- 广播变量 (一般写在算子的后面即可)
- 使用 withBroadcastSet(data, “name”) 这个方法即可, name变量代表了获取该广播变量的名称
- 使用广播变量
- 使用方法主要是通过 RichFunction, 在 对应的 open( )方法中, 可以根据名称来获取对应的广播变量, 只需要一次获取, 就可以一直使用了, 具体方法如下:
dataSet.map(new RichMapFunction<String, String>() {
List bc;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
// 2. 获取广播变量
this.bc = getRuntimeContext().getBroadcastVariable(“broadcastData”);
}
@Override
public String map(String s) throws Exception {
return s;
}
// 1. 将需要用的变量广播出去 (这一步可以写在后面)
}).withBroadcastSet(broadcastData, “broadcastData”).print();
下面以一个获取用户年龄的例子来演示一个常见的使用案例:
broadcastData 是一个包含用户 (姓名, 年龄) 的数据表
需要在另外一个算子中通过姓名查找年龄, 那么就需要把上表广播
public class BroadcastExample { public static void main(String[] args) throws Exception { ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(4); // 创建需要广播的 数据集 (name, age) Tuple2<String, Integer> john = new Tuple2<>("john", 23); Tuple2<String, Integer> tom = new Tuple2<>("tom", 24); Tuple2<String, Integer> shiny = new Tuple2<>("shiny", 22); DataSource<Tuple2<String, Integer>> broadcastData = env.fromElements(john, tom, shiny); // 新建一个dataset -> d1, 设置并行度为4 // 此时 d1 是无法访问 broadcastData 的数据的, 因为两个dataset可能不在一个节点或者slot中, 所以 flink 是不允许去访问的 DataSet<String> d1 = env.fromElements("john", "tom", "shiny").setParallelism(4); // 使用 RichMapFunction, 在open() 方法中拿到广播变量 d1.map(new RichMapFunction<String, String>() { List<Tuple2<String, Integer>> bc; HashMap<String, Integer> map = new HashMap<>(); @Override public void open(Configuration parameters) throws Exception { super.open(parameters); this.bc = getRuntimeContext().getBroadcastVariable("broadcastData"); for (Tuple2<String, Integer> tp : bc) { this.map.put(tp.f0, tp.f1); } } @Override public String map(String s) throws Exception { Integer age = this.map.get(s); return s + "->" + age; } }).withBroadcastSet(broadcastData, "broadcastData").print(); } }
