背景

對(duì)于web網(wǎng)站，我們一般會(huì)有這樣的需求，實(shí)時(shí)的計(jì)算出來(lái)當(dāng)天網(wǎng)站的uv，盡可能快的展示出來(lái)。今天我們就講一下基于java的set集合做一下實(shí)時(shí)uv的統(tǒng)計(jì)。

簡(jiǎn)易需求：

• 實(shí)時(shí)計(jì)算出當(dāng)天零點(diǎn)截止到當(dāng)前時(shí)間各個(gè)端(android,ios,h6)下的uv
• 每秒鐘更新一次統(tǒng)計(jì)結(jié)果

案例講解

模擬source

首先我們模擬生成一下最簡(jiǎn)單的數(shù)據(jù)，生成一個(gè)flink的二元組Tuple2.分別表示分類和用戶id

 public static class MySource implements SourceFunction<Tuple2<String,Integer>>{
  private volatile boolean isRunning = true;
  String category[] = {"Android", "IOS", "H5"};
  @Override
  public void run(SourceContext<Tuple2<String,Integer>> ctx) throws Exception{
   while (isRunning){
    Thread.sleep(10);
    //具體是哪個(gè)端的用戶
    String type = category[(int) (Math.random() * (category.length))];
    //隨機(jī)生成10000以內(nèi)的int類型數(shù)據(jù)作為userid
    int userid = (int) (Math.random() * 10000);
    ctx.collect(Tuple2.of(type, userid));
   }
  }
  @Override
  public void cancel(){
   isRunning = false;
  }
 }

定義窗口

接下來(lái)我們定義一個(gè)周期是一天的滑動(dòng)窗口，因?yàn)槲覀円棵腌娸敵龃翱诘臄?shù)據(jù)，所以我們緊接著窗口定義了一個(gè)1秒的觸發(fā)器。

DataStream<Tuple2<String,Integer>> dataStream = env.addSource(new MySource());
    dataStream.keyBy(0).window(TumblingProcessingTimeWindows.of(Time.days(1), Time.hours(-8)))
              .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(1)))
              .aggregate(new MyAggregate(),new WindowResult())
              .print();

自定義聚合算子

接下來(lái)我們自定義一個(gè)聚合算子來(lái)實(shí)現(xiàn)該功能。

對(duì)于聚合算子的理解可以參考這個(gè)文章：

https://mp.weixin.qq.com/s/ZCWexNGzhSchRpxipa1x-g

 public static class MyAggregate
   implements AggregateFunction<Tuple2<String,Integer>,Set<Integer>,Integer>{
  @Override
  public Set<Integer> createAccumulator(){
   return new HashSet<>();
  }
  @Override
  public Set<Integer> add(Tuple2<String,Integer> value, Set<Integer> accumulator){
   accumulator.add(value.f1);
   return accumulator;
  }
  @Override
  public Integer getResult(Set<Integer> accumulator){
   return accumulator.size();
  }
  @Override
  public Set<Integer> merge(Set<Integer> a, Set<Integer> b){
   a.addAll(b);
   return a;
  }
 }

處理輸出結(jié)果

我們這里將結(jié)果輸出到控制臺(tái)，實(shí)際的生產(chǎn)中我們可以將數(shù)據(jù)寫入redis或者h(yuǎn)base等。

1> Result{, dateTime='2020-06-21 19:23:30'type='IOS', uv=136}
2> Result{, dateTime='2020-06-21 19:23:30'type='Android', uv=150}
1> Result{, dateTime='2020-06-21 19:23:30'type='H5', uv=134}
1> Result{, dateTime='2020-06-21 19:23:31'type='IOS', uv=164}
2> Result{, dateTime='2020-06-21 19:23:31'type='Android', uv=177}
1> Result{, dateTime='2020-06-21 19:23:31'type='H5', uv=167}
2> Result{, dateTime='2020-06-21 19:23:32'type='Android', uv=205}
1> Result{, dateTime='2020-06-21 19:23:32'type='IOS', uv=193}
1> Result{, dateTime='2020-06-21 19:23:32'type='H5', uv=198}