devbox
羊村懒王
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Flink的累加器和广播变量、广播流、分布式缓存_flink 中accumulator

作者：羊村懒王 | 2024-06-09 23:20:38

赞

踩

flink 中accumulator

1、Accumulator累加器 


Accumulator即累加器，与Mapreduce counter的应用场景差不多，都能很好地观察task在运行期间的数据变化。可以在Flink job任务中的算子函数中使用累加器，但是只能在任务执行结束后才能获得累加器的最终结果。Counter是一个具体的累加器（Accumulator）实现，常用的Counter有IntCounter，LongCounter和DoubleCounter。

用法：

  1.     1：创建累加器
  2.     private IntCounter numLines = new IntCounter();
  3.     2：注册累加器
  4.     getRuntimeContext().addAccumulator("num-lines",this.numLines);
  5.     3:使用累加器
  6.     this.numLines.add(1);
  7.     4:获取累加器的结果
  8.     myJobExcutionResult.getAccumulatorResult("num-lines")

 案列：统计map算子处理数据的条数

  1. package Flink_API;
  2.  
  3. import org.apache.flink.api.common.accumulators.IntCounter;
  4. import org.apache.flink.api.common.functions.RichMapFunction;
  5. import org.apache.flink.api.java.DataSet;
  6. import org.apache.flink.api.java.ExecutionEnvironment;
  7. import org.apache.flink.api.java.operators.DataSource;
  8. import org.apache.flink.api.java.operators.MapOperator;
  9. import org.apache.flink.configuration.Configuration;
  10.  
  11. /**
  12.  * 统计一下map函数处理了多少条数据
  13.  */
  14. public class BatchCounterTest {
  15.     public static void main(String[] args) throws Exception {
  16.  
  17.         ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
  18.  
  19.         DataSource<String> dataSource=env.fromElements("1","2","3","4","5");
  20.  
  21.         DataSet<String> map = dataSource.map(new RichMapFunction<String, String>() {
  22.             //            1：创建累加器
  23.             private IntCounter numLines = new IntCounter();
  24.  
  25.             @Override
  26.             public void open(Configuration parameters) throws Exception {
  27.                 //注册累加器
  28.                 getRuntimeContext().addAccumulator("num-lines", numLines);
  29.             }
  30.  
  31.             @Override
  32.             public String map(String s) throws Exception {
  33.                //使用累加器
  34.                 numLines.add(1);
  35.                 return s;
  36.             }
  37.         }).setParallelism(5);
  38.         map.print();
  39.         env.execute("BatchCounterTest");
  40.     }
  41. }

2、广播变量：是通过广播将广播变量分发到taskmanager中进行处理

  1. 广播变量的使用步骤：
  2.     1、初始化数据
  3.     DataSet<Integer> toBroadcast = env.fromElements(1,2,3);
  4.     2、广播数据（即注册数据，那个算子用，就在那个算子后面进行注册）
  5.     算子.withBroadcastSet(toBroadcast,"broadcastSetName");
  6.     3、获取数据
  7.     Collection<Integer> broadcastSet = getRuntimeContext().getBroadcastVariable("broadcastSetName");


实例程序：Flink从数据园中静静可以获取到用户的性命，最终需要将用户的性命和年龄信息打印出来。

  1. package Flink_API;
  2.  
  3. import org.apache.flink.api.common.functions.MapFunction;
  4. import org.apache.flink.api.common.functions.RichMapFunction;
  5. import org.apache.flink.api.java.DataSet;
  6. import org.apache.flink.api.java.ExecutionEnvironment;
  7. import org.apache.flink.api.java.operators.DataSource;
  8. import org.apache.flink.api.java.tuple.Tuple2;
  9. import org.apache.flink.configuration.Configuration;
  10.  
  11. import java.util.ArrayList;
  12. import java.util.HashMap;
  13. import java.util.List;
  14.  
  15. public class BatchBroadcastTest {
  16.     public static void main(String[] args){
  17.         //获取Flink的运行环境
  18.         ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
  19.         //准备需要的广播数据
  20.         ArrayList<Tuple2<String,Integer>> broadData=new ArrayList<>();
  21.         broadData.add(new Tuple2<>("wtt",29));
  22.         broadData.add(new Tuple2<>("lidong",30));
  23.         broadData.add(new Tuple2<>("hengda",40));
  24.         DataSource<Tuple2<String,Integer>> tupleData=env.fromCollection(broadData);
  26.         //处理需要广播的数据，将数据集转换成Map类型，Map中的key就是用户的性命，value就是用户年龄。
  27.         DataSet<HashMap<String,Integer>> toBroadCast = tupleData.map(new MapFunction<Tuple2<String, Integer>, HashMap<String,Integer>>() {
  28.             @Override
  29.             public HashMap<String, Integer> map(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
  30.                 HashMap<String,Integer> hashMap=new HashMap<>();
  31.                 hashMap.put(stringIntegerTuple2.f0,stringIntegerTuple2.f1);
  32.                 return hashMap;
  33.             }
  34.         }).setParallelism(3);//到此，广播的数据已经准备好了
  35.  
  36.         //注意：在这里使用RichMapFunction获取广播变量
  37.         //数据源单纯的姓名信息
  38.         DataSource<String> nameDataSource = env.fromElements("wtt","lidong","hengda");
  39.  
  40.         DataSet<String> data=nameDataSource.map(new RichMapFunction<String, String>() {
  41.  
  42.             List<HashMap<String,Integer>> broadCastMap=new ArrayList<HashMap<String,Integer>>();
  43.             HashMap<String,Integer> allMap=new HashMap<String,Integer>();
  44.  
  45.             /**
  46.              * 1、类似MR当中的setup方法，只会执行一次
  47.              * 2、可以在这里进行一些初始化操作
  48.              * 3、可以在open方法当中获取广播变量
  49.              */
  50.             @Override
  51.             public void open(Configuration parameters) throws Exception {
  52.                 super.open(parameters);
  53.                 //获取广播数据
  54.                 broadCastMap = getRuntimeContext().getBroadcastVariable("toBroadCastMapName");
  55.                 for(HashMap map:broadCastMap){
  56.                     allMap.putAll(map);//最终保存的格式就是{"name":"age"}
  57.  
  58.                 }
  59.             }
  60.             /**
  61.              *
  62.              *每次条用map方法从allMap中获取数据即可
  63.              */
  64.             @Override
  65.             public String map(String s) throws Exception {
  66.                 return s;
  67.             }
  68.         });
  69.     }
  70. }

3、广播流：批处理当中就是广播变量，流处理当中就是广播流

  1. package Flink_API;
  2.  
  3. import org.apache.flink.api.common.serialization.SimpleStringSchema;
  4. import org.apache.flink.api.common.state.BroadcastState;
  5. import org.apache.flink.api.common.state.MapStateDescriptor;
  6. import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
  7. import org.apache.flink.streaming.api.TimeCharacteristic;
  8. import org.apache.flink.streaming.api.datastream.BroadcastStream;
  9. import org.apache.flink.streaming.api.datastream.DataStream;
  10. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  11. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  12. import org.apache.flink.streaming.api.functions.ProcessFunction;
  13. import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
  14. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
  15. import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
  16. import org.apache.flink.util.Collector;
  17.  
  18. import java.io.Serializable;
  19. import java.util.Properties;
  20.  
  21. //广播流
  22. public class FlinkBroadcastStream {
  23.  
  24.     public static void main(String[] args) throws Exception {
  25.         //创建运行环境
  26.         StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
  27.         //Flink是以数据自带的时间戳字段为准
  28.         env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  29.         //设置并行度
  30.         env.setParallelism(1);
  31.  
  32.          //1、获取第一个流，获取用户的浏览信息
  33.         DataStream<UserBrowseLog> browseStream = getUserBrowseDataStream(env);
  34.         //获取用户的黑名单流信息
  35.         //2、获取用户的点击信息
  36.         DataStream<BlackUser> blackUserDataStream = getUserBlackUserDataStream(env);
  37.  
  38.         //1定义一个MapStateDescriptor来描述我们要广播的数据的格式
  39.         MapStateDescriptor<String,BlackUser> descriptor=new MapStateDescriptor<String, BlackUser>("userBlackList",String.class,BlackUser.class);
  40.  
  41.         //2将其中的配置数据源注册成广播流
  42.         BroadcastStream<BlackUser> broadcastStream = blackUserDataStream.broadcast(descriptor);
  43.  
  44.  
  45.         //3通过connect连接主流和广播流
  46.         DataStream<UserBrowseLog> filterDataStream = browseStream.connect(broadcastStream)
  47.                 .process(new BroadcastProcessFunction<UserBrowseLog, BlackUser, UserBrowseLog>(){
  48.                     @Override
  49.                     public void processElement(UserBrowseLog value, ReadOnlyContext readOnlyContext, Collector<UserBrowseLog> collector) throws Exception {
  50.                           //从广播中获取对应的key的value
  51.                         ReadOnlyBroadcastState<String,BlackUser> broadcastState=readOnlyContext.getBroadcastState(descriptor);
  52.                         BlackUser blackUser=broadcastState.get(value.userID);
  53.                         if(blackUser !=null){
  54.                             System.out.print("用户"+value.userID + "在黑名单中，过滤掉该用户的浏览信息");
  55.                         }else{
  56.                             collector.collect(value);
  57.                         }
  58.                     }
  59.  
  60.                     @Override
  61.                     public void processBroadcastElement(BlackUser value, Context context, Collector<UserBrowseLog> collector) throws Exception {
  62.                         //实时更新广播流当中的数据
  63.                         BroadcastState<String,BlackUser> broadcastState=context.getBroadcastState(descriptor);
  64.                         broadcastState.put(value.userID,value);
  65.                         System.out.print("------------------>广播流当前的数据是：---------------->");
  66.                         System.out.print(broadcastState);
  67.                     }
  68.                 });
  69.         filterDataStream.print();
  70.         env.execute("FlinkBroadcastStream");
  71.     }
  72.  
  73.     private static DataStream<UserBrowseLog> getUserBrowseDataStream(StreamExecutionEnvironment env) {
  74.         Properties consumerProperties = new Properties();
  75.         consumerProperties.setProperty("bootstrap.severs","page01:9001");
  76.         consumerProperties.setProperty("grop.id","browsegroup");
  77.  
  78.         DataStreamSource<String> dataStreamSource=env.addSource(new FlinkKafkaConsumer010<String>("browse_topic", (KeyedDeserializationSchema<String>) new SimpleStringSchema(),consumerProperties));
  79.  
  80.         DataStream<UserBrowseLog> processData=dataStreamSource.process(new ProcessFunction<String, UserBrowseLog>() {
  81.             @Override
  82.             public void processElement(String s, Context context, Collector<UserBrowseLog> collector) throws Exception {
  83.                 try{
  84.                     UserBrowseLog browseLog = com.alibaba.fastjson.JSON.parseObject(s, UserBrowseLog.class);
  85.                     if(browseLog !=null){
  86.                         collector.collect(browseLog);
  87.                     }
  88.                 }catch(Exception e){
  89.                     System.out.print("解析Json——UserBrowseLog异常："+e.getMessage());
  90.                 }
  91.             }
  92.         });
  93.         //设置watermark
  94.         return processData;
  95.     }
  96.  
  97.     private static DataStream<BlackUser> getUserBlackUserDataStream(StreamExecutionEnvironment env) {
  98.         Properties consumerProperties = new Properties();
  99.         consumerProperties.setProperty("bootstrap.severs","page01:9002");
  100.         consumerProperties.setProperty("grop.id","browsegroup");
  101.  
  102.         DataStreamSource<String> dataStreamSource=env.addSource(new FlinkKafkaConsumer010<String>("browse_topic", (KeyedDeserializationSchema<String>) new SimpleStringSchema(),consumerProperties));
  103.  
  104.         DataStream<BlackUser> processData=dataStreamSource.process(new ProcessFunction<String, BlackUser>() {
  105.             @Override
  106.             public void processElement(String s, Context context, Collector<BlackUser> collector) throws Exception {
  107.                 try{
  108.                     BlackUser blackUser = com.alibaba.fastjson.JSON.parseObject(s, BlackUser.class);
  109.                     if(blackUser !=null){
  110.                         collector.collect(blackUser);
  111.                     }
  112.                 }catch(Exception e){
  113.                     System.out.print("解析Json——UserBrowseLog异常："+e.getMessage());
  114.                 }
  115.             }
  116.         });
  117.         return processData;
  118.     }
  119.     //定义用户黑名单的配置信息
  120.     public static class BlackUser implements Serializable{
  121.         private String userID;
  122.         private String userName;
  123.         public BlackUser(){
  124.  
  125.         }
  126.  
  127.         public BlackUser(String userID, String userName) {
  128.             this.userID = userID;
  129.             this.userName = userName;
  130.         }
  131.  
  132.         public String getUserID() {
  133.             return userID;
  134.         }
  135.  
  136.         public void setUserID(String userID) {
  137.             this.userID = userID;
  138.         }
  139.  
  140.         public String getUserName() {
  141.             return userName;
  142.         }
  143.  
  144.         public void setUserName(String userName) {
  145.             this.userName = userName;
  146.         }
  147.     }
  148.     //浏览类
  149.     public static class UserBrowseLog implements Serializable {
  150.         private String userID;
  151.         private String eventTime;
  152.         private String eventType;
  153.         private String productID;
  154.         private Integer productPrice;
  155.  
  156.         public String getUserID() {
  157.             return userID;
  158.         }
  159.  
  160.         public void setUserID(String userID) {
  161.             this.userID = userID;
  162.         }
  163.  
  164.         public String getEventTime() {
  165.             return eventTime;
  166.         }
  167.  
  168.         public void setEventTime(String eventTime) {
  169.             this.eventTime = eventTime;
  170.         }
  171.  
  172.         public String getEventType() {
  173.             return eventType;
  174.         }
  175.  
  176.         public void setEventType(String eventType) {
  177.             this.eventType = eventType;
  178.         }
  179.  
  180.         public String getProductID() {
  181.             return productID;
  182.         }
  183.  
  184.         public void setProductID(String productID) {
  185.             this.productID = productID;
  186.         }
  187.  
  188.         public Integer getProductPrice() {
  189.             return productPrice;
  190.         }
  191.  
  192.         public void setProductPrice(Integer productPrice) {
  193.             this.productPrice = productPrice;
  194.         }
  195.  
  196.         @Override
  197.         public String toString() {
  198.             return "UserBrowseLog{" +
  199.                     "userID='" + userID + '\'' +
  200.                     ", eventTime='" + eventTime + '\'' +
  201.                     ", eventType='" + eventType + '\'' +
  202.                     ", productID='" + productID + '\'' +
  203.                     ", productPrice=" + productPrice +
  204.                     '}';
  205.         }
  206.     }
  207. }

4、Flink分布式缓存Distributed Cache

  • Flink提供了一个分布式缓存，类似于hadoop，可以使用户在并行函数中很方便的读取本地文件，并把它放在taskmanager节点中，防止task重复拉取。
  • 此缓存的工作机制如下：程序注册一个文件或者目录(本地或者远程文件系统，例如hdfs或者s3)，通过ExecutionEnvironment注册缓存文件并为它起一个名称。当程序执行，Flink自动将文件或者目录复制到所有taskmanager节点的本地文件系统，仅会执行一次。用户可以通过这个指定的名称查找文件或者目录，然后从taskmanager节点的本地文件系统访问它

注册: 

  1. //获取运行环境
  2. ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
  3. //1：注册一个文件,可以使用hdfs上的文件 也可以是本地文件进行测试
  4. env.registerCachedFile("/Users/wangzhiwu/WorkSpace/quickstart/text","a.txt");

使用： 

 File myFile = getRuntimeContext().getDistributedCache().getFile("a.text");

 a.text文件


hello flink hello FLINK

完整代码：

  1. public class DisCacheTest {
  2.     public static void main(String[] args) throws Exception{
  3.         //获取运行环境
  4.         ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
  5.         //1：注册一个文件,可以使用hdfs上的文件 也可以是本地文件进行测试
  6.       
  7.         env.registerCachedFile("/Users/wangzhiwu/WorkSpace/quickstart/text","a.txt");
  8.         DataSource<String> data = env.fromElements("a", "b", "c", "d");
  9.         DataSet<String> result = data.map(new RichMapFunction<String, String>() {
  10.             private ArrayList<String> dataList = new ArrayList<String>();
  11.             @Override
  12.             public void open(Configuration parameters) throws Exception {
  13.                 super.open(parameters);
  14.                 //2：使用文件
  15.                 File myFile = getRuntimeContext().getDistributedCache().getFile("a.txt");
  16.                 List<String> lines = FileUtils.readLines(myFile);
  17.                 for (String line : lines) {
  18.                     this.dataList.add(line);
  19.                     System.err.println("分布式缓存为:" + line);
  20.                 }
  21.             }
  22.             @Override
  23.             public String map(String value) throws Exception {
  24.                 //在这里就可以使用dataList
  25.                 System.err.println("使用datalist：" + dataList + "------------" +value);
  26.                 //业务逻辑
  27.                 return dataList +"：" +  value;
  28.             }
  29.         });
  30.         result.printToErr();
  31.     }
  32. }//

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/羊村懒王/article/detail/696198
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号