欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

Flink Broadcast 与 Accumulators 应用案例实战-Flink牛刀小试

程序员文章站 2022-03-08 09:10:45
...

版权声明:本套技术专栏是作者(秦凯新)平时工作的总结和升华,通过从真实商业环境抽取案例进行总结和分享,并给出商业应用的调优建议和集群环境容量规划等内容,请持续关注本套博客。版权声明:禁止转载,欢迎学习。QQ邮箱地址:[email protected],如有任何问题,可随时联系。

1.1 Broadcast 真假曹操

  • DataStreaming Broadcast (元素广播):元素广播,重复处理

    • 把元素广播给所有的分区,数据会被重复处理,类似于storm中的allGrouping
    • 使用技巧:dataStream.broadcast()
  • Flink Broadcast(广播变量)

    • 广播变量创建后,它可以运行在集群中的任何function上,而不需要多次传递给集群节点。
      另外需要记住,不应该修改广播变量,这样才能确保每个节点获取到的值都是一致的。

    • 一句话解释,可以理解为是一个公共的共享变量,我们可以把一个dataset 数据集广播出去,然后不同的task在节点上都能够获取到,这个数据在每个节点上只会存
      在一份。

    • 如果不使用broadcast,则在每个节点中的每个task中都需要拷贝一份dataset数据集,比较浪费内存(也就是一个节点中可能会存在多份dataset数据)。

    • 用法如下:

        1:初始化数据
        DataSet<Integer> toBroadcast = env.fromElements(1, 2, 3)
        2:广播数据
        .withBroadcastSet(toBroadcast, "broadcastSetName");
        3:获取数据
        Collection<Integer> broadcastSet = getRuntimeContext().getBroadcastVariable("broadcastSetName");
        
        注意:
        1:广播出去的变量存在于每个节点的内存中,所以这个数据集不能太大。因为广播出去的数据,会常驻内存,除非程序执行结束
        2:广播变量在初始化广播出去以后不支持修改,这样才能保证每个节点的数据都是一致的。
      

2 元素广播案例实战

2.1 实现元素的重复广播,设置source的并行度为1

public class StreamingDemoWithMyNoPralalleSourceBroadcast {

    public static void main(String[] args) throws Exception {
        //获取Flink的运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        //获取数据源
        DataStreamSource<Long> text = env.addSource(new MyNoParalleSource()).setParallelism(1);//注意:针对此source,并行度只能设置为1

        DataStream<Long> num = text.broadcast().map(new MapFunction<Long, Long>() {
            @Override
            public Long map(Long value) throws Exception {
                long id = Thread.currentThread().getId();
                System.out.println("线程id:"+id+",接收到数据:" + value);
                return value;
            }
        });

        //每2秒钟处理一次数据
        DataStream<Long> sum = num.timeWindowAll(Time.seconds(2)).sum(0);

        //打印结果
        sum.print().setParallelism(1);

        String jobName = StreamingDemoWithMyNoPralalleSourceBroadcast.class.getSimpleName();
        env.execute(jobName);
    }
}

2.2 自定义接收器MyNoParalleSource

    public class MyNoParalleSource implements SourceFunction<Long>{

        private long count = 1L;
    
        private boolean isRunning = true;
    
        /**
         * 主要的方法
         * 启动一个source
         * 大部分情况下,都需要在这个run方法中实现一个循环,这样就可以循环产生数据了
         *
         * @param ctx
         * @throws Exception
         */
        @Override
        public void run(SourceContext<Long> ctx) throws Exception {
            while(isRunning){
                ctx.collect(count);
                count++;
                //每秒产生一条数据
                Thread.sleep(1000);
            }
        }
    
        /**
         * 取消一个cancel的时候会调用的方法
         *
         */
        @Override
        public void cancel() {
            isRunning = false;
        }
}

2.3 结果展示

发现整个Map元素别处理了4次:

线程id:44,接收到数据:1
线程id:46,接收到数据:1
线程id:42,接收到数据:1
线程id:48,接收到数据:1
4

3 广播变量

3.1 第一步:封装DataSet,调用withBroadcastSet。

3.2 第二步:getRuntimeContext().getBroadcastVariable,获得广播变量

3.3 第三步:RichMapFunction中执行获得广播变量的逻辑

public class BatchDemoBroadcast {

    public static void main(String[] args) throws Exception{

        //获取运行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        //1:准备需要广播的数据
        ArrayList<Tuple2<String, Integer>> broadData = new ArrayList<>();
        broadData.add(new Tuple2<>("zs",18));
        broadData.add(new Tuple2<>("ls",20));
        broadData.add(new Tuple2<>("ww",17));
        DataSet<Tuple2<String, Integer>> tupleData = env.fromCollection(broadData);


        //1.1:处理需要广播的数据,把数据集转换成map类型,map中的key就是用户姓名,value就是用户年龄
        DataSet<HashMap<String, Integer>> toBroadcast = tupleData.map(new MapFunction<Tuple2<String, Integer>, HashMap<String, Integer>>() {
            @Override
            public HashMap<String, Integer> map(Tuple2<String, Integer> value) throws Exception {
                HashMap<String, Integer> res = new HashMap<>();
                res.put(value.f0, value.f1);
                return res;
            }
        });

        //源数据
        DataSource<String> data = env.fromElements("zs", "ls", "ww");

        //注意:在这里需要使用到RichMapFunction获取广播变量
        DataSet<String> result = data.map(new RichMapFunction<String, String>() {

            List<HashMap<String, Integer>> broadCastMap = new ArrayList<HashMap<String, Integer>>();
            HashMap<String, Integer> allMap = new HashMap<String, Integer>();

            /**
             * 这个方法只会执行一次
             * 可以在这里实现一些初始化的功能
             *
             * 所以,就可以在open方法中获取广播变量数据
             *
             */
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                //3:获取广播数据
                this.broadCastMap = getRuntimeContext().getBroadcastVariable("broadCastMapName");
                for (HashMap map : broadCastMap) {
                    allMap.putAll(map);
                }

            }

            @Override
            public String map(String value) throws Exception {
                Integer age = allMap.get(value);
                return value + "," + age;
            }
        }).withBroadcastSet(toBroadcast, "broadCastMapName");//2:执行广播数据的操作
        result.print();
    }
}

3.4 结果展示

    zs,18
    ls,20
    ww,17

总结

简单成文,方便Flink整体体系构成,感谢Github FLink 源码作者,让我学到很多东西。辛苦成文,各自珍惜,谢谢!

版权声明:本套技术专栏是作者(秦凯新)平时工作的总结和升华,通过从真实商业环境抽取案例进行总结和分享,并给出商业应用的调优建议和集群环境容量规划等内容,请持续关注本套博客。版权声明:禁止转载,欢迎学习。QQ邮箱地址:[email protected],如有任何问题,可随时联系。

秦凯新 于深圳 20181608

相关标签: Flink