flink 广播变量&&累加器&&计数器_flink map获取累加值

##用法

##1.初始化数据

DataSet<Integer> toBroadcast =env.fromElements(1,2,3)

##2.广播数据

   .withBroadcastSet(toBoradcast,”broadcastsetname”);

##3.获取数据

   Collection<Integer> broadcastSet=getRuntimeContext().getBroadcastVariable(“broadcastsetname”);

###注意

广播出去的变量存在于每个节点的内存中，所以这个 数据集不能太大，因为广播出去的数据，会常驻内存，除非程序执行结束

广播变量在初始化广播出去之后不支持修改，这样才能保持一致性。

public static void main(String[] args) throws Exception{
  ExecutionEnvironment env=ExecutionEnvironment.getExecutionEnvironment();

  //准备数据
  ArrayList<Tuple2<String,Integer>> broadcast=new ArrayList<>();
  broadcast.add(new Tuple2<>("zhangsan",29));
  broadcast.add(new Tuple2<>("lisi",45));
  broadcast.add(new Tuple2<>("wangwu",98));
  DataSet<Tuple2<String,Integer>> ds=env.fromCollection(broadcast);
  //处理要广播的数据，吧数据集转换成map类型，map中的key就是用户姓名，value就是用户年龄
  DataSet<HashMap<String,Integer>>tobct= ds.map(new MapFunction<Tuple2<String,Integer>, HashMap<String,Integer>>() {
    @Override
    public HashMap<String, Integer> map(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
      HashMap<String ,Integer> res=new HashMap<>();
      res.put(stringIntegerTuple2.f0,stringIntegerTuple2.f1);
      return res;
    }
  });


  //源数据
  DataSource<String> data=env.fromElements("zhangsan","wangwu","lisi");

  DataSet<String> result=data.map(new RichMapFunction<String, String>() {
    //这个方法只会执行一次可以实现初始化的功能
    List<HashMap<String,Integer>>  broadcastMap=new ArrayList<>();
    HashMap<String,Integer> allmap=new HashMap<>();
    @Override
    public void open(Configuration parameters) throws Exception {
      //可以获取广播数据
      super.open(parameters);
       this.broadcastMap=getRuntimeContext().getBroadcastVariable("ds");
       for (HashMap map :broadcastMap){
         allmap.putAll(map);
       }
    }
    //来一次数据处理一次,参数表示用户名
    @Override
    public String map(String s) throws Exception {
      Integer age=allmap.get(s);
      return s+" "+age;
    }
  }).withBroadcastSet(tobct,"ds");//执行广播数据操作;
  result.print();
}

IntCounter，LongCounter和DoubleCounter

用法

1. 创建累加器

Private IntCounter numlines=new IntCounter();

1. 注册累加器

getRuntimContext().addAccumulator(“num-lines”,this.numlines)

1. 使用累加器

This.numlines.add(1)

1. 获取累加器结果

myJobExcutionResult.getAccumulatorResutl(“num-lines”)

不适用累加器，直接自定义sum进行计数

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

  ExecutionEnvironment env= ExecutionEnvironment.getExecutionEnvironment();

  

  DataSource<String> data=env.fromElements("a","b","c","d");

  

  DataSet<String> result=data.map(new RichMapFunction<String, String>() {

    //1.创建累加器

    private IntCounter numlines=new IntCounter();

  

   // int sum=0;

    @Override

    public void open(Configuration parameters) throws Exception {

      super.open(parameters);

      //注册累加器

      getRuntimeContext().addAccumulator("num-lines",this.numlines);

    }

  

    @Override

    public String map(String s) throws Exception {

      //如果并行度为1,使用普通的累加求和，结果正确，但是设置多个并行度，则普通的累加器求和结果不对

      //sum++;

      //System.out.println("sum: "+sum);

      //3.使用累加器

      this.numlines.add(1);

      return s;

    }

  });

  result.writeAsText("d:\\counter.text");

  JobExecutionResult jobResult=env.execute("counter");

  int num=jobResult.getAccumulatorResult("num-lines");

  System.out.println("num: "+num);

  }

累加器是可以在不同任务中对同一个变量进行累加操作

Broadcast允许程序员将一个只读的变量缓存在每台机器上，而不用在任务间传递变量，广播变量可以共享，但是不能修改。

###用法

1. 注册一个文件

EnvregisterCachedFile(“hdfs://master:9000/data”,”hdfsFlie”);

1. 访问数据

FilemyFile =getRuntimeContext().getDistributedCache().getFile(“hdfsFile”);

如果出现以下异常

flink Could not find a file system implementation for scheme 'hdfs'. The scheme is not directly supply…..

请在pom文件中加入

<dependency>

   <groupId>org.apache.flink</groupId>

   <artifactId>flink-shaded-hadoop2</artifactId>

   <version>${flink.version}</version>

</dependency>

并启动hadoop集群

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

  ExecutionEnvironment env=ExecutionEnvironment.getExecutionEnvironment();

  

  //1、注册一个文件

  env.registerCachedFile("hdfs://master:9000/data/fof/input/friend.txt","hdfs");

  

  DataSource<String> data= env.fromElements("a","b","c","d");

  DataSet<String> result = data.map(new RichMapFunction<String, String>() {

  

    private ArrayList<String> alist = new ArrayList();

  

    @Override

    public void open(Configuration parameters) throws Exception {

      super.open(parameters);

  

      //使用文件

      File myFile = getRuntimeContext().getDistributedCache().getFile("hdfs");

      List<String> list = FileUtils.readLines(myFile);

      for (String line : list) {

        this.alist.add(line);

        System.out.println(line);

      }

    }

  

    @Override

    public String map(String s) throws Exception {

      return s;

    }

  });

  result.print();

  }