亚洲激情专区-91九色丨porny丨老师-久久久久久久女国产乱让韩-国产精品午夜小视频观看

溫馨提示×

溫馨提示×

您好,登錄后才能下訂單哦!

密碼登錄×
登錄注冊×
其他方式登錄
點擊 登錄注冊 即表示同意《億速云用戶服務條款》

flink中如何使用set實時計算當天網站uv

發布時間:2021-11-09 18:53:13 來源:億速云 閱讀:263 作者:柒染 欄目:大數據

flink中如何使用set實時計算當天網站uv,針對這個問題,這篇文章詳細介紹了相對應的分析和解答,希望可以幫助更多想解決這個問題的小伙伴找到更簡單易行的方法。

背景

對于web網站,我們一般會有這樣的需求,實時的計算出來當天網站的uv,盡可能快的展示出來。今天我們就講一下基于java的set集合做一下實時uv的統計。

簡易需求:

  • 實時計算出當天零點截止到當前時間各個端(android,ios,h6)下的uv
  • 每秒鐘更新一次統計結果
 

案例講解

 

模擬source

首先我們模擬生成一下最簡單的數據,生成一個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))];
    //隨機生成10000以內的int類型數據作為userid
    int userid = (int) (Math.random() * 10000);
    ctx.collect(Tuple2.of(type, userid));
   }
  }
  @Override
  public void cancel(){
   isRunning = false;
  }
 }

   

定義窗口

接下來我們定義一個周期是一天的滑動窗口,因為我們要每秒鐘輸出窗口的數據,所以我們緊接著窗口定義了一個1秒的觸發器。


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

 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等。


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}

關于flink中如何使用set實時計算當天網站uv問題的解答就分享到這里了,希望以上內容可以對大家有一定的幫助,如果你還有很多疑惑沒有解開,可以關注億速云行業資訊頻道了解更多相關知識。

向AI問一下細節

免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。

AI

土默特右旗| 天全县| 安顺市| 澄城县| 金秀| 广安市| 中牟县| 石柱| 浦江县| 顺昌县| 南郑县| 安岳县| 革吉县| 抚顺县| 九龙坡区| 平定县| 晋江市| 呼图壁县| 宜章县| 永川市| 邹城市| 肇东市| 汉阴县| 富顺县| 兴化市| 含山县| 晋州市| 邢台市| 个旧市| 柘城县| 阜城县| 盐源县| 商河县| 旬邑县| 洪洞县| 新丰县| 杭锦后旗| 莱西市| 梁河县| 扬中市| 岑溪市|