美文网首页flink简单使用教程
flink使用11-了解broadcast的用法

flink使用11-了解broadcast的用法

作者: CheckChe | 来源:发表于2019-11-09 22:12 被阅读0次

    在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();
        }
    }
    

    相关文章

      网友评论

        本文标题:flink使用11-了解broadcast的用法

        本文链接:https://www.haomeiwen.com/subject/xafobctx.html