flink中的聚合算子是什么

前言

flink中的一個接口org.apache.flink.api.common.functions.AggregateFunction，這個類可以接在window流之后，做窗口內(nèi)的統(tǒng)計計算。

注意：除了這個接口AggregateFunction，flink中還有一個抽象類AggregateFunction：org.apache.flink.table.functions.AggregateFunction，大家不要把這個弄混淆了，接口AggregateFunction我們可以理解為flink中的一個算子，和MapFunction、FlatMapFunction等是同級別的，而抽象類AggregateFunction是用于用戶自定義聚合函數(shù)的，和max、min之類的函數(shù)是同級的。

原理解析

比如我們想實現(xiàn)一個類似sql的功能:

select TUMBLE_START(proctime,INTERVAL '2' SECOND)  as starttime,user,count(*) from logs group by user,TUMBLE(proctime,INTERVAL '2' SECOND)

這個sql就是來統(tǒng)計一下每兩秒鐘的滑動窗口內(nèi)每個人出現(xiàn)的次數(shù)，今天我們就以這個簡單的sql的功能為例講解一下flink的aggregate算子，其實就是我們用程序來實現(xiàn)這個sql的功能。

首先看一下聚合函數(shù)的接口:

@PublicEvolving
public interface AggregateFunction<IN, ACC, OUT> extends Function, Serializable {
 ACC createAccumulator();
 ACC add(IN value, ACC accumulator);
 ACC merge(ACC a, ACC b);
 OUT getResult(ACC accumulator);
}

這個接口AggregateFunction里面有4個方法，我們分別來講解一下。

1. AggregateFunction這個類是一個泛型類，這里面有三個參數(shù)，IN, ACC, OUT。IN就是聚合函數(shù)的輸入類型，ACC是存儲中間結(jié)果的類型，OUT是聚合函數(shù)的輸出類型。
2. createAccumulator    
   這個方法首先要創(chuàng)建一個累加器，要進行一些初始化的工作，比如我們要進行count計數(shù)操作，就要給累加器一個初始值。
3. add    
   add方法就是我們要做聚合的時候的核心邏輯，比如我們做count累加，其實就是來一個數(shù)，然后就加一。    
   類似上面的sql的邏輯，我們在寫業(yè)務(wù)邏輯的時候，可以這么想，進入這方法數(shù)的數(shù)據(jù)都是屬于某一個用戶的，系統(tǒng)在調(diào)用這個方法之前會先進行hash分組，然后不同的用戶會重復(fù)調(diào)用這個方法。所以這個函數(shù)的入?yún)⑹荌N類型，返回值是ACC類型
4. merge    
   因為flink是一個分布式計算框架，可能計算是分布在很多節(jié)點上同時進行的，比如上述的add操作，可能同一個用戶在不同的節(jié)點上分別調(diào)用了add方法在本地節(jié)點對本地的數(shù)據(jù)進行了聚合操作，但是我們要的是整個結(jié)果，整個時候，我們就需要把每個用戶各個節(jié)點上的聚合結(jié)果merge一下，整個merge方法就是做這個工作的，所以它的入?yún)⒑统鰠⒌念愋投际侵虚g結(jié)果類型ACC。
5. getResult    
   這個方法就是將每個用戶最后聚合的結(jié)果經(jīng)過處理之后，按照OUT的類型返回，返回的結(jié)果也就是聚合函數(shù)的輸出結(jié)果了。

實例講解 

自定義source

首先我們自定義source生成用戶的信息

 public static class MySource implements SourceFunction<Tuple2<String,Long>>{

  private volatile boolean isRunning = true;

  String userids[] = {
    "4760858d-2bec-483c-a535-291de04b2247", "67088699-d4f4-43f2-913c-481bff8a2dc5",
    "72f7b6a8-e1a9-49b4-9a0b-770c41e01bfb", "dfa27cb6-bd94-4bc0-a90b-f7beeb9faa8b",
    "aabbaa50-72f4-495c-b3a1-70383ee9d6a4", "3218bbb9-5874-4d37-a82d-3e35e52d1702",
    "3ebfb9602ac07779||3ebfe9612a007979", "aec20d52-c2eb-4436-b121-c29ad4097f6c",
    "e7e896cd939685d7||e7e8e6c1930689d7", "a4b1e1db-55ef-4d9d-b9d2-18393c5f59ee"
  };

  @Override
  public void run(SourceContext<Tuple2<String,Long>> ctx) throws Exception{
   while (isRunning){
    Thread.sleep(10);
    String userid = userids[(int) (Math.random() * (userids.length - 1))];
    ctx.collect(Tuple2.of(userid, System.currentTimeMillis()));
   }
  }

  @Override
  public void cancel(){
   isRunning = false;
  }
 }
  
  

自定義聚合函數(shù)

 public static class CountAggregate
   implements AggregateFunction<Tuple2<String,Long>,Integer,Integer>{

  @Override
  public Integer createAccumulator(){
   return 0;
  }

  @Override
  public Integer add(Tuple2<String,Long> value, Integer accumulator){
   return ++accumulator;
  }

  @Override
  public Integer getResult(Integer accumulator){
   return accumulator;
  }

  @Override
  public Integer merge(Integer a, Integer b){
   return a + b;
  }
 }
  
  

自定義結(jié)果輸出函數(shù)

 /**
  * 這個是為了將聚合結(jié)果輸出
  */
 public static class WindowResult
   implements WindowFunction<Integer,Tuple3<String,Date,Integer>,Tuple,TimeWindow>{

  @Override
  public void apply(
    Tuple key,
    TimeWindow window,
    Iterable<Integer> input,
    Collector<Tuple3<String,Date,Integer>> out) throws Exception{

   String k = ((Tuple1<String>) key).f0;
   long windowStart = window.getStart();
   int result = input.iterator().next();
   out.collect(Tuple3.of(k, new Date(windowStart), result));

  }
 }
  
  

主流程

 final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  DataStream<Tuple2<String,Long>> dataStream = env.addSource(new MySource());

  dataStream.keyBy(0).window(TumblingProcessingTimeWindows.of(Time.seconds(2)))
            .aggregate(new CountAggregate(), new WindowResult()
            ).print();

  env.execute();