前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >flink实战教程-使用set实时计算当天网站uv

flink实战教程-使用set实时计算当天网站uv

作者头像
大数据技术与应用实战
发布2020-09-15 14:33:27
1K0
发布2020-09-15 14:33:27
举报
文章被收录于专栏:大数据技术与应用实战

背景

对于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

本文参与 腾讯云自媒体同步曝光计划,分享自微信公众号。
原始发表:2020-06-22,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 大数据技术与应用实战 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 背景
  • 案例讲解
    • 模拟source
      • 定义窗口
        • 自定义聚合算子
          • 处理输出结果
          相关产品与服务
          大数据
          全栈大数据产品,面向海量数据场景,帮助您 “智理无数,心中有数”!
          领券
          问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档