Flink系列之Flink中Broadcast和Counter整理和实战

title: Flink系列

6.1 理论

Flink 的批处理 和 Spark 的批处理&＃xff0c;都支持两个非常好的特性&＃xff1a; 广播变量 &＃43; 累加器

广播变量允许编程人员在每台机器上保持1个只读的缓存变量&＃xff0c;而不是传送变量的副本给tasks&＃xff0c;广播变量创建后&＃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;这样才能保证每个节点的数据都是一致的。


6.2 案例

package com.aa.flinkjava.broadcast;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
/**
* &＃64;Author AA
* &＃64;Date 2022/2/24 19:37
* &＃64;Project bigdatapre
* &＃64;Package com.aa.flinkjava.broadcast
* Flink BroadCast 测试
* 在这里做一个join的连接实现
*/
public class FlinkBroadCastDemo {
public static void main(String[] args) throws Exception {
//1、获取运行环境
ExecutionEnvironment executionEnvironment &＃61; ExecutionEnvironment.getExecutionEnvironment();
//2、造数据
ArrayList<Tuple2<String,Integer>> list &＃61; new ArrayList<>();
list.add(new Tuple2<>("zhangsan",20));
list.add(new Tuple2<>("lisi",21));
list.add(new Tuple2<>("wangwu",22));
//3、读取造的数据
DataSource<Tuple2<String, Integer>> dataSource &＃61; executionEnvironment.fromCollection(list);
dataSource.print("dataSource : ");
//4、帮tuple2转化为hashmap 。 map中的key是用户姓名&＃xff0c;value是用户年龄
// DataSet>的数据类型可以直接修饰强制转换。
DataSet<HashMap<String, Integer>> toBroadcast &＃61; dataSource.map(new MapFunction<Tuple2<String, Integer>, HashMap<String, Integer>>() {
&＃64;Override
public HashMap<String, Integer> map(Tuple2<String, Integer> tuple2) throws Exception {
HashMap<String, Integer> hashMap &＃61; new HashMap<>();
hashMap.put(tuple2.f0,tuple2.f1);
return hashMap;
}
});
//5、再造一份 join 使用的数据
DataSource<String> data2 &＃61; executionEnvironment.fromElements("zhangsan", "lisi", "wangwu");
data2.print("data2 : ");
//6、执行广播数据的一些操作
// 下面这个DataSet类型也是强制转换的的
DataSet<String> result &＃61; data2.map(new RichMapFunction<String, String>() {
List<HashMap<String, Integer>> broadCastMap &＃61; new ArrayList<HashMap<String, Integer>>();
HashMap<String, Integer> allMap &＃61; new HashMap<String, Integer>();
&＃64;Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
this.broadCastMap &＃61; getRuntimeContext().getBroadcastVariable("bdMapName");
for (HashMap map : broadCastMap) {
allMap.putAll(map);
}
}
/**
* &＃64;param s s是data2中间的一个一个的元素&＃xff0c;其实就是"zhangsan", "lisi", "wangwu" 这些值
* 根据 name("zhangsan", "lisi", "wangwu") 去广播变量中匹配获取相应的年龄
* &＃64;return
* &＃64;throws Exception
*/
&＃64;Override
public String map(String s) throws Exception {
Integer age &＃61; allMap.get(s);
return s &＃43; "," &＃43; age; //输出拼接的结果
}
}).withBroadcastSet(toBroadcast, "bdMapName");
//7、打印输出
result.print();
}
}

七、Flink Counter 编程实战

7.1 理论

​ Accumulator 即累加器&＃xff0c;与 Mapreduce Counter 的应用场景差不多&＃xff0c;都能很好地观察 Task 在运行期间的数据变化。可以在 Flink job 任务中的算子函数中操作累加器&＃xff0c;但是只能在任务执行结束之后才能获得累加器的最终结果。

​ Counter 是一个具体的累加器 (Accumulator) 实现&＃xff1a;IntCounter, LongCounter 和 DoubleCounter

用法&＃xff1a;

// 1、创建累加器
private IntCounter numlines &＃61; new IntCounter();
// 2、注册累加器
getRuntimeContext().addAccumulator("num", this.numLines);
// 3、使用累加器
this.numlines.add(1);
// 4、获取累加器的结果
myJobExecutionResult.getAccumulatorResult("num")


7.2 案例

package com.aa.flinkjava.counter;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.accumulators.IntCounter;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import java.util.ArrayList;
/**
* &＃64;Author AA
* &＃64;Date 2022/2/25 14:27
* &＃64;Project bigdatapre
* &＃64;Package com.aa.flinkjava.counter
* Flink 累加器 示例
* 统计输入数据源的流入数据的次数。
*/
public class FlinkCounterDemo {
public static void main(String[] args) throws Exception {
//1、获取运行环境
ExecutionEnvironment executionEnvironment &＃61; ExecutionEnvironment.getExecutionEnvironment();
executionEnvironment.setParallelism(3);
//2、读取造的数据
DataSource<String> dataSource &＃61; executionEnvironment.fromElements("a", "b", "c", "d", "e", "f", "g", "h", "i", "j");
//3、定义一点逻辑&＃xff0c;给累加器放进去
MapOperator<String, String> result &＃61; dataSource.map(new RichMapFunction<String, String>() {
//3-1 创建累加器对象
private IntCounter numlines &＃61; new IntCounter();
&＃64;Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
//3-2 需要注册累加器
/*
在逻辑上来说&＃xff0c;相当于在这个 application应用内部定义了一个变量 num 用来做统计。
但是&＃xff0c;物理上&＃xff0c;其实这个 num 变量是由分散在所有 Task 内部的 numlines 组成的。
一个 num 包含了很多个 numlines。其实最终拿到的结果&＃xff0c;就是把所有 Task 中的 numlines 加起来&＃xff0c;就是 num 的值。
*/
this.getRuntimeContext().addAccumulator("num", this.numlines);
}
&＃64;Override
public String map(String s) throws Exception {
//另外注意&＃xff0c;可能有小伙伴觉得可以在这里定义普通变量统计也行&＃xff0c;
// 注意&＃xff1a;若并行度为1&＃xff0c;使用普通的累加求和也可以&＃xff0c;但是设置多个并行度&＃xff0c;则普通的累加求和结果就不准啦。
//每运行一次就 向累加器中 添加1
this.numlines.add(1);
return s; //这里没有做什么逻辑&＃xff0c;就是给来的数据原样输出了。但是上面统计了累加次数了。
}
});
//4、给结果输出出去
result.writeAsText("D:\\flinkcount3");
//5、执行
JobExecutionResult jobExecutionResult &＃61; executionEnvironment.execute();
//6、看看累加器的结果
Integer num &＃61; jobExecutionResult.getAccumulatorResult("num");
System.out.println("累加器的输出的结果是&＃xff1a; " &＃43; num);
}
}


声明&＃xff1a;
        文章中代码及相关语句为自己根据相应理解编写&＃xff0c;文章中出现的相关图片为自己实践中的截图和相关技术对应的图片&＃xff0c;若有相关异议&＃xff0c;请联系删除。感谢。转载请注明出处&＃xff0c;感谢。

B站&＃xff1a; https://space.bilibili.com/1523287361 点击打开链接
微博地址&＃xff1a; http://weibo.com/luoyepiaoxue2014 点击打开链接