FlinkBroadcast广播变量应用案例实战Flink牛刀小试

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

1.1 Broadcast 真假曹操

DataStreaming Broadcast (元素广播)&＃xff1a;元素广播&＃xff0c;重复处理
- 把元素广播给所有的分区&＃xff0c;数据会被重复处理&＃xff0c;类似于storm中的allGrouping
- 使用技巧&＃xff1a;dataStream.broadcast()
Flink Broadcast(广播变量)
- 广播变量创建后&＃xff0c;它可以运行在集群中的任何function上&＃xff0c;而不需要多次传递给集群节点。另外需要记住&＃xff0c;不应该修改广播变量&＃xff0c;这样才能确保每个节点获取到的值都是一致的。
- 一句话解释&＃xff0c;可以理解为是一个公共的共享变量&＃xff0c;我们可以把一个dataset 数据集广播出去&＃xff0c;然后不同的task在节点上都能够获取到&＃xff0c;这个数据在每个节点上只会存在一份。
- 如果不使用broadcast&＃xff0c;则在每个节点中的每个task中都需要拷贝一份dataset数据集&＃xff0c;比较浪费内存(也就是一个节点中可能会存在多份dataset数据)。
- 用法如下&＃xff1a;
  
  1&＃xff1a;初始化数据DataSet toBroadcast &＃61; env.fromElements(1, 2, 3)2&＃xff1a;广播数据.withBroadcastSet(toBroadcast, "broadcastSetName");3&＃xff1a;获取数据Collection broadcastSet &＃61; getRuntimeContext().getBroadcastVariable("broadcastSetName");注意&＃xff1a;1&＃xff1a;广播出去的变量存在于每个节点的内存中&＃xff0c;所以这个数据集不能太大。因为广播出去的数据&＃xff0c;会常驻内存&＃xff0c;除非程序执行结束2&＃xff1a;广播变量在初始化广播出去以后不支持修改&＃xff0c;这样才能保证每个节点的数据都是一致的。

2 元素广播案例实战

2.1 实现元素的重复广播&＃xff0c;设置source的并行度为1

public class StreamingDemoWithMyNoPralalleSourceBroadcast {public static void main(String[] args) throws Exception {//获取Flink的运行环境StreamExecutionEnvironment env &＃61; StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(4);//获取数据源DataStreamSource text &＃61; env.addSource(new MyNoParalleSource()).setParallelism(1);//注意&＃xff1a;针对此source&＃xff0c;并行度只能设置为1DataStream num &＃61; text.broadcast().map(new MapFunction() {&＃64;Overridepublic Long map(Long value) throws Exception {long id &＃61; Thread.currentThread().getId();System.out.println("线程id&＃xff1a;"&＃43;id&＃43;",接收到数据&＃xff1a;" &＃43; value);return value;}});//每2秒钟处理一次数据DataStream sum &＃61; num.timeWindowAll(Time.seconds(2)).sum(0);//打印结果sum.print().setParallelism(1);String jobName &＃61; StreamingDemoWithMyNoPralalleSourceBroadcast.class.getSimpleName();env.execute(jobName);} }

2.2 自定义接收器MyNoParalleSource

public class MyNoParalleSource implements SourceFunction{private long count &＃61; 1L;private boolean isRunning &＃61; true;/*** 主要的方法* 启动一个source* 大部分情况下&＃xff0c;都需要在这个run方法中实现一个循环&＃xff0c;这样就可以循环产生数据了** &＃64;param ctx* &＃64;throws Exception*/&＃64;Overridepublic void run(SourceContext ctx) throws Exception {while(isRunning){ctx.collect(count);count&＃43;&＃43;;//每秒产生一条数据Thread.sleep(1000);}}/*** 取消一个cancel的时候会调用的方法**/&＃64;Overridepublic void cancel() {isRunning &＃61; false;} }

2.3 结果展示

发现整个Map元素别处理了4次&＃xff1a;

线程id&＃xff1a;44,接收到数据&＃xff1a;1 线程id&＃xff1a;46,接收到数据&＃xff1a;1 线程id&＃xff1a;42,接收到数据&＃xff1a;1 线程id&＃xff1a;48,接收到数据&＃xff1a;1 4

3 广播变量

3.1 第一步&＃xff1a;封装DataSet&＃xff0c;调用withBroadcastSet。

3.2 第二步&＃xff1a;getRuntimeContext().getBroadcastVariable&＃xff0c;获得广播变量

3.3 第三步&＃xff1a;RichMapFunction中执行获得广播变量的逻辑

public class BatchDemoBroadcast {public static void main(String[] args) throws Exception{//获取运行环境ExecutionEnvironment env &＃61; ExecutionEnvironment.getExecutionEnvironment();//1&＃xff1a;准备需要广播的数据ArrayList> broadData &＃61; new ArrayList<>();broadData.add(new Tuple2<>("zs",18));broadData.add(new Tuple2<>("ls",20));broadData.add(new Tuple2<>("ww",17));DataSet> tupleData &＃61; env.fromCollection(broadData);//1.1:处理需要广播的数据,把数据集转换成map类型&＃xff0c;map中的key就是用户姓名&＃xff0c;value就是用户年龄DataSet> toBroadcast &＃61; tupleData.map(new MapFunction, HashMap>() {&＃64;Overridepublic HashMap map(Tuple2 value) throws Exception {HashMap res &＃61; new HashMap<>();res.put(value.f0, value.f1);return res;}});//源数据DataSource data &＃61; env.fromElements("zs", "ls", "ww");//注意&＃xff1a;在这里需要使用到RichMapFunction获取广播变量DataSet result &＃61; data.map(new RichMapFunction() {List> broadCastMap &＃61; new ArrayList>();HashMap allMap &＃61; new HashMap();/*** 这个方法只会执行一次* 可以在这里实现一些初始化的功能** 所以&＃xff0c;就可以在open方法中获取广播变量数据**/&＃64;Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);//3:获取广播数据this.broadCastMap &＃61; getRuntimeContext().getBroadcastVariable("broadCastMapName");for (HashMap map : broadCastMap) {allMap.putAll(map);}}&＃64;Overridepublic String map(String value) throws Exception {Integer age &＃61; allMap.get(value);return value &＃43; "," &＃43; age;}}).withBroadcastSet(toBroadcast, "broadCastMapName");//2&＃xff1a;执行广播数据的操作result.print();} }