在Flink中，同一个算子可能存在若干个不同的并行实例，计算过程可能不在同一个Slot中进行，不同算子之间更是如此，因此不同算子的计算数据之间不能像Java数组之间一样互相访问，而广播变量Broadcast便是解决这种情况的. 在 flink 中, 针对某一个算子需要使用公共变量的情况下, 就可以把对应的数据给广播出去, 这样在所有的节点中都可以使用了. 典型的代码结构如下所示:
在一个算子中使用广播变量主要有两个步骤:
-
广播变量 (一般写在算子的后面即可)
使用 withBroadcastSet(data, "name") 这个方法即可, name变量代表了获取该广播变量的名称
-
使用广播变量
使用方法主要是通过 RichFunction, 在 对应的 open( )方法中, 可以根据名称来获取对应的广播变量, 只需要一次获取, 就可以一直使用了, 具体方法如下:
dataSet.map(new RichMapFunction<String, String>() {
List<Integer> 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();
}
}
