flink使用11-了解broadcast的用法

在Flink中，同一个算子可能存在若干个不同的并行实例，计算过程可能不在同一个Slot中进行，不同算子之间更是如此，因此不同算子的计算数据之间不能像Java数组之间一样互相访问，而广播变量Broadcast便是解决这种情况的. 在 flink 中, 针对某一个算子需要使用公共变量的情况下, 就可以把对应的数据给广播出去, 这样在所有的节点中都可以使用了. 典型的代码结构如下所示:

在一个算子中使用广播变量主要有两个步骤:

1. 广播变量 (一般写在算子的后面即可)

   使用 withBroadcastSet(data, "name") 这个方法即可, name变量代表了获取该广播变量的名称

2. 使用广播变量

   使用方法主要是通过 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();
    }
}