flink中如何使用set實(shí)時計算當(dāng)天網(wǎng)站uv

背景

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

簡易需求：

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

案例講解

模擬source

首先我們模擬生成一下最簡單的數(shù)據(jù)，生成一個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);
    //具體是哪個端的用戶
    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;
  }
 }

定義窗口

接下來我們定義一個周期是一天的滑動窗口，因?yàn)槲覀円棵腌娸敵龃翱诘臄?shù)據(jù)，所以我們緊接著窗口定義了一個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();

自定義聚合算子

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

對于聚合算子的理解可以參考這個文章：

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é)果輸出到控制臺，實(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}