背景
对于web网站，我们一般会有这样的需求，实时的计算出来当天网站的uv，尽可能快的展示出来。今天我们就讲一下基于java的set集合做一下实时uv的统计。
简易需求：
- 实时计算出当天零点截止到当前时间各个端(android,ios,h5)下的uv
- 每秒钟更新一次统计结果
案例讲解
模拟source
首先我们模拟生成一下最简单的数据，生成一个flink的二元组Tuple2.分别表示分类和用户id
代码语言：javascript复制
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))];
//随机生成10000以内的int类型数据作为userid
int userid = (int) (Math.random() * 10000);
ctx.collect(Tuple2.of(type, userid));
}
}
@Override
public void cancel(){
isRunning = false;
}
}
定义窗口
接下来我们定义一个周期是一天的滑动窗口，因为我们要每秒钟输出窗口的数据，所以我们紧接着窗口定义了一个1秒的触发器。
代码语言：javascript复制
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();
自定义聚合算子
接下来我们自定义一个聚合算子来实现该功能。
对于聚合算子的理解可以参考这个文章：
https://mp.weixin.qq.com/s/ZCWexNGzhSchRpxipa1x-g
代码语言：javascript复制 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;
}
}
处理输出结果
我们这里将结果输出到控制台，实际的生产中我们可以将数据写入redis或者hbase等。
代码语言：javascript复制
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}
完整代码请参考 https://github.com/zhangjun0x01/bigdata-examples/blob/master/flink/src/main/java/windows/RealTimePvUv_Set.java
