Flink实时计算Pv、Uv的几种方法

本文转载自微信公众号「Java大数据与数据仓库」，作者柯少爷。转载本文请联系Java大数据与数据仓库公众号。

实时统计pv、uv是再常见不过的大数据统计需求了，前面出过一篇SparkStreaming实时统计pv,uv的案例，这里用Flink实时计算pv,uv。

我们需要统计不同数据类型每天的pv，uv情况,并且有如下要求.

  • 每秒钟要输出最新的统计结果;
  • 程序永远跑着不会停,所以要定期清理内存里的过时数据;
  • 收到的消息里的时间字段并不是按照顺序严格递增的,所以要有一定的容错机制;
  • 访问uv并不一定每秒钟都会变化,重复输出对IO是巨大的浪费,所以要在uv变更时在一秒内输出结果,未变更时不输出;

Flink数据流上的类型和操作

DataStream是flink流处理最核心的数据结构，其它的各种流都可以直接或者间接通过DataStream来完成相互转换，一些常用的流直接的转换关系如图：

可以看出，DataStream可以与KeyedStream相互转换，KeyedStream可以转换为WindowedStream，DataStream不能直接转换为WindowedStream，WindowedStream可以直接转换为DataStream。各种流之间虽然不能相互直接转换，但是都可以通过先转换为DataStream，再转换为其它流的方法来实现。

在这个计算pv,uv的需求中就主要用到DataStream、KeyedStream以及WindowedStream这些数据结构。

这里需要用到window和watermark，使用窗口把数据按天分割，使用watermark可以通过“水位”来定期清理窗口外的迟到数据，起到清理内存的作用。

业务代码

我们的数据是json类型的，含有date,version,guid这3个字段，在实时统计pv,uv这个功能中，其它字段可以直接丢掉，当然了在离线数据仓库中，所有有含义的业务字段都是要保留到hive当中的。其它相关概念就不说了，会专门介绍，这里直接上代码吧。

 
 
 
 
    
  
  
  
  
  1.  
    2.   
  
  
  
  2.   
  
  
  
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
    4.   
  
  
  
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> 
    5.   
  
  
  
  5.     4.0.0 
    6.   
  
  
  
  6.  
    7.   
  
  
  
  7.     com.ddxygq 
    8.   
  
  
  
  8.     bigdata 
    9.   
  
  
  
  9.     1.0-SNAPSHOT 
    10.   
  
  
  
  10.  
    11.   
  
  
  
  11.      
    12.   
  
  
  
  12.         2.11.8 
    13.   
  
  
  
  13.         1.7.0 
    14.   
  
  
  
  14.         bigdata 
    15.   
  
  
  
  15.      
    16.   
  
  
  
  16.  
    17.   
  
  
  
  17.      
    18.   
  
  
  
  18.          
    19.   
  
  
  
  19.             org.apache.flink 
    20.   
  
  
  
  20.             flink-scala_2.11 
    21.   
  
  
  
  21.             {flink.version} 
    22.   
  
  
  
  22.    
    23.   
  
  
  
  23.          
    24.   
  
  
  
  24.             org.apache.flink 
    25.   
  
  
  
  25.             flink-streaming-scala_2.11 
    26.   
  
  
  
  26.             flink.version 
    27.   
  
  
  
  27.    
    28.   
  
  
  
  28.    
    29.   
  
  
  
  29.          
    30.   
  
  
  
  30.             org.apache.flink 
    31.   
  
  
  
  31.             flink-streaming-java_2.11 
    32.   
  
  
  
  32.             {flink.version} 
    33.   
  
  
  
  33.          
    34.   
  
  
  
  34.          
    35.   
  
  
  
  35.          
    36.   
  
  
  
  36.             org.apache.flink 
    37.   
  
  
  
  37.             flink-connector-kafka-0.10_2.11 
    38.   
  
  
  
  38.             flink.version 
    39.   
  
  
  
  39.    
    40.   
  
  
  
  40.  
    41.   
  
  
  
  41.      
    42.   
  
  
  
  42.          
    43.   
  
  
  
  43.          
    44.   
  
  
  
  44.         basedir/src/test−−>{pkg.name} 
    45.   
  
  
  
  45.         src/main/java 
    46.   
  
  
  
  46.          
    47.   
  
  
  
  47.              
    48.   
  
  
  
  48.                 src/main/resources 
    49.   
  
  
  
  49.                  
    50.   
  
  
  
  50.                     *.properties 
    51.   
  
  
  
  51.                     *.xml 
    52.   
  
  
  
  52.                  
    53.   
  
  
  
  53.                 false 
    54.   
  
  
  
  54.              
    55.   
  
  
  
  55.          
    56.   
  
  
  
  56.          
    57.   
  
  
  
  57.              
    58.   
  
  
  
  58.              
    59.   
  
  
  
  59.                 org.apache.maven.plugins 
    60.   
  
  
  
  60.                 maven-surefire-plugin 
    61.   
  
  
  
  61.                  
    62.   
  
  
  
  62.                     true 
    63.   
  
  
  
  63.                  
    64.   
  
  
  
  64.              
    65.   
  
  
  
  65.              
    66.   
  
  
  
  66.              
    67.   
  
  
  
  67.                 org.scala-tools 
    68.   
  
  
  
  68.                 maven-scala-plugin 
    69.   
  
  
  
  69.                 2.15.2 
    70.   
  
  
  
  70.                  
    71.   
  
  
  
  71.                      
    72.   
  
  
  
  72.                          
    73.   
  
  
  
  73.                             compile 
    74.   
  
  
  
  74.                             testCompile 
    75.   
  
  
  
  75.                          
    76.   
  
  
  
  76.                      
    77.   
  
  
  
  77.                  
    78.   
  
  
  
  78.              
    79.   
  
  
  
  79.          
    80.   
  
  
  
  80.      
    81.   
  
  
  
  81.  
    82.

主要代码，主要使用scala开发：

 
 
 
 
    
  
  
  
  
  1. package com.ddxygq.bigdata.flink.streaming.pvuv 
    2.   
  
  
  
  2.  
    3.   
  
  
  
  3. import java.util.Properties 
    4.   
  
  
  
  4.  
    5.   
  
  
  
  5. import com.alibaba.fastjson.JSON 
    6.   
  
  
  
  6. import org.apache.flink.runtime.state.filesystem.FsStateBackend 
    7.   
  
  
  
  7. import org.apache.flink.streaming.api.CheckpointingMode 
    8.   
  
  
  
  8. import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor 
    9.   
  
  
  
  9. import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment} 
    10.   
  
  
  
  10. import org.apache.flink.streaming.api.windowing.time.Time 
    11.   
  
  
  
  11. import org.apache.flink.streaming.api.windowing.triggers.ContinuousProcessingTimeTrigger 
    12.   
  
  
  
  12. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010 
    13.   
  
  
  
  13. import org.apache.flink.streaming.util.serialization.SimpleStringSchema 
    14.   
  
  
  
  14. import org.apache.flink.streaming.api.scala.extensions._ 
    15.   
  
  
  
  15. import org.apache.flink.api.scala._ 
    16.   
  
  
  
  16.  
    17.   
  
  
  
  17. /** 
    18.   
  
  
  
  18.   * @ Author: keguang 
    19.   
  
  
  
  19.   * @ Date: 2019/3/18 17:34 
    20.   
  
  
  
  20.   * @ version: v1.0.0 
    21.   
  
  
  
  21.   * @ description:  
    22.   
  
  
  
  22.   */ 
    23.   
  
  
  
  23. object PvUvCount { 
    24.   
  
  
  
  24.   def main(args: Array[String]): Unit = { 
    25.   
  
  
  
  25.     val env = StreamExecutionEnvironment.getExecutionEnvironment 
    26.   
  
  
  
  26.  
    27.   
  
  
  
  27.     // 容错 
    28.   
  
  
  
  28.     env.enableCheckpointing(5000) 
    29.   
  
  
  
  29.     env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE) 
    30.   
  
  
  
  30.     env.setStateBackend(new FsStateBackend("file:///D:/space/IJ/bigdata/src/main/scala/com/ddxygq/bigdata/flink/checkpoint/flink/tagApp")) 
    31.   
  
  
  
  31.  
    32.   
  
  
  
  32.     // kafka 配置 
    33.   
  
  
  
  33.     val ZOOKEEPER_HOST = "hadoop01:2181,hadoop02:2181,hadoop03:2181" 
    34.   
  
  
  
  34.     val KAFKA_BROKERS = "hadoop01:9092,hadoop02:9092,hadoop03:9092" 
    35.   
  
  
  
  35.     val TRANSACTION_GROUP = "flink-count" 
    36.   
  
  
  
  36.     val TOPIC_NAME = "flink" 
    37.   
  
  
  
  37.     val kafkaProps = new Properties() 
    38.   
  
  
  
  38.     kafkaProps.setProperty("zookeeper.connect", ZOOKEEPER_HOST) 
    39.   
  
  
  
  39.     kafkaProps.setProperty("bootstrap.servers", KAFKA_BROKERS) 
    40.   
  
  
  
  40.     kafkaProps.setProperty("group.id", TRANSACTION_GROUP) 
    41.   
  
  
  
  41.  
    42.   
  
  
  
  42.     // watrmark 允许数据延迟时间 
    43.   
  
  
  
  43.     val MaxOutOfOrderness = 86400 * 1000L 
    44.   
  
  
  
  44.      
    45.   
  
  
  
  45.     // 消费kafka数据 
    46.   
  
  
  
  46.     val streamData: DataStream[(String, String, String)] = env.addSource( 
    47.   
  
  
  
  47.       new FlinkKafkaConsumer010[String](TOPIC_NAME, new SimpleStringSchema(), kafkaProps) 
    48.   
  
  
  
  48.     ).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[String](Time.milliseconds(MaxOutOfOrderness)) { 
    49.   
  
  
  
  49.       override def extractTimestamp(element: String): Long = { 
    50.   
  
  
  
  50.         val t = JSON.parseObject(element) 
    51.   
  
  
  
  51.         val time = JSON.parseObject(JSON.parseObject(t.getString("message")).getString("decrypted_data")).getString("time") 
    52.   
  
  
  
  52.         time.toLong 
    53.   
  
  
  
  53.       } 
    54.   
  
  
  
  54.     }).map(x => { 
    55.   
  
  
  
  55.       var date = "error" 
    56.   
  
  
  
  56.       var guid = "error" 
    57.   
  
  
  
  57.       var helperversion = "error" 
    58.   
  
  
  
  58.       try { 
    59.   
  
  
  
  59.         val messageJsonObject = JSON.parseObject(JSON.parseObject(x).getString("message")) 
    60.   
  
  
  
  60.         val datetime = messageJsonObject.getString("time") 
    61.   
  
  
  
  61.         date = datetime.split(" ")(0) 
    62.   
  
  
  
  62.         // hour = datetime.split(" ")(1).substring(0, 2) 
    63.   
  
  
  
  63.         val decrypted_data_string = messageJsonObject.getString("decrypted_data") 
    64.   
  
  
  
  64.         if (!"".equals(decrypted_data_string)) { 
    65.   
  
  
  
  65.           val decrypted_data = JSON.parseObject(decrypted_data_string) 
    66.   
  
  
  
  66.           guid = decrypted_data.getString("guid").trim 
    67.   
  
  
  
  67.           helperversion = decrypted_data.getString("helperversion") 
    68.   
  
  
  
  68.         } 
    69.   
  
  
  
  69.       } catch { 
    70.   
  
  
  
  70.         case e: Exception => { 
    71.   
  
  
  
  71.           println(e) 
    72.   
  
  
  
  72.         } 
    73.   
  
  
  
  73.       } 
    74.   
  
  
  
  74.       (date, helperversion, guid) 
    75.   
  
  
  
  75.     }) 
    76.   
  
  
  
  76.     // 这上面是设置watermark并解析json部分 
    77.   
  
  
  
  77.     // 聚合窗口中的数据，可以研究下applyWith这个方法和OnWindowedStream这个类 
    78.   
  
  
  
  78.     val resultStream = streamData.keyBy(x => { 
    79.   
  
  
  
  79.       x._1 + x._2 
    80.   
  
  
  
  80.     }).timeWindow(Time.days(1)) 
    81.   
  
  
  
  81.       .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(1))) 
    82.   
  
  
  
  82.       .applyWith(("", List.empty[Int], Set.empty[Int], 0L, 0L))( 
    83.   
  
  
  
  83.         foldFunction = { 
    84.   
  
  
  
  84.           case ((_, list, set, _, 0), item) => { 
    85.   
  
  
  
  85.             val date = item._1 
    86.   
  
  
  
  86.             val helperversion = item._2 
    87.   
  
  
  
  87.             val guid = item._3 
    88.   
  
  
  
  88.             (date + "_" + helperversion, guid.hashCode +: list, set + guid.hashCode, 0L, 0L) 
    89.   
  
  
  
  89.           } 
    90.   
  
  
  
  90.         } 
    91.   
  
  
  
  91.         , windowFunction = { 
    92.   
  
  
  
  92.           case (key, window, result) => { 
    93.   
  
  
  
  93.             result.map { 
    94.   
  
  
  
  94.               case (leixing, list, set, _, _) => { 
    95.   
  
  
  
  95.                 (leixing, list.size, set.size, window.getStart, window.getEnd) 
    96.   
  
  
  
  96.               } 
    97.   
  
  
  
  97.             } 
    98.   
  
  
  
  98.           } 
    99.   
  
  
  
  99.         } 
    100.   
  
  
  
  100.       ).keyBy(0) 
    101.   
  
  
  
  101.       .flatMapWithState[(String, Int, Int, Long, Long),(Int, Int)]{ 
    102.   
  
  
  
  102.       case ((key, numpv, numuv, begin, end), curr) => 
    103.   
  
  
  
  103.  
    104.   
  
  
  
  104.         curr match { 
    105.   
  
  
  
  105.           case Some(numCurr) if numCurr == (numuv, numpv) => 
    106.   
  
  
  
  106.             (Seq.empty, Some((numuv, numpv))) //如果之前已经有相同的数据,则返回空结果 
    107.   
  
  
  
  107.           case _ => 
    108.   
  
  
  
  108.             (Seq((key, numpv, numuv, begin, end)), Some((numuv, numpv))) 
    109.   
  
  
  
  109.         } 
    110.   
  
  
  
  110.     } 
    111.   
  
  
  
  111.  
    112.   
  
  
  
  112.     // 最终结果 
    113.   
  
  
  
  113.     val resultedStream = resultStream.map(x => { 
    114.   
  
  
  
  114.       val keys = x._1.split("_") 
    115.   
  
  
  
  115.       val date = keys(0) 
    116.   
  
  
  
  116.       val helperversion = keys(1) 
    117.   
  
  
  
  117.       (date, helperversion, x._2, x._3) 
    118.   
  
  
  
  118.     }) 
    119.   
  
  
  
  119.  
    120.   
  
  
  
  120.     resultedStream.print() 
    121.   
  
  
  
  121.     env.execute("PvUvCount") 
    122.   
  
  
  
  122.  
    123.   
  
  
  
  123.   } 
    124.   
  
  
  
  124. } 
    125.

使用List集合的size保存pv,使用Set集合的size保存uv，从而达到实时统计pv,uv的目的。

这里用了几个关键的函数：

applyWith：里面需要的参数，初始状态变量，和foldFunction ，windowFunction ;

存在的问题

显然，当数据量很大的时候，这个List集合和Set集合会很大，并且这里的pv是否可以不用List来存储，而是通过一个状态变量，不断做累加，对应操作就是更新状态来完成。

改进版

使用了一个计数器来存储pv的值。

 
 
 
 
    
  
  
  
  
  1. packagecom.ddxygq.bigdata.flink.streaming.pvuv 
    2.   
  
  
  
  2.  
    3.   
  
  
  
  3. import java.util.Properties 
    4.   
  
  
  
  4.  
    5.   
  
  
  
  5. import com.alibaba.fastjson.JSON 
    6.   
  
  
  
  6. import org.apache.flink.api.common.accumulators.IntCounter 
    7.   
  
  
  
  7. import org.apache.flink.runtime.state.filesystem.FsStateBackend 
    8.   
  
  
  
  8. import org.apache.flink.streaming.api.CheckpointingMode 
    9.   
  
  
  
  9. import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor 
    10.   
  
  
  
  10. import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment} 
    11.   
  
  
  
  11. import org.apache.flink.streaming.api.windowing.time.Time 
    12.   
  
  
  
  12. import org.apache.flink.streaming.api.windowing.triggers.ContinuousProcessingTimeTrigger 
    13.   
  
  
  
  13. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010 
    14.   
  
  
  
  14. import org.apache.flink.streaming.util.serialization.SimpleStringSchema 
    15.   
  
  
  
  15. import org.apache.flink.streaming.api.scala.extensions._ 
    16.   
  
  
  
  16. import org.apache.flink.api.scala._ 
    17.   
  
  
  
  17. import org.apache.flink.core.fs.FileSystem 
    18.   
  
  
  
  18.  
    19.   
  
  
  
  19. object PvUv2 { 
    20.   
  
  
  
  20.   def main(args: Array[String]): Unit = { 
    21.   
  
  
  
  21.     val env = StreamExecutionEnvironment.getExecutionEnvironment 
    22.   
  
  
  
  22.  
    23.   
  
  
  
  23.     // 容错 
    24.   
  
  
  
  24.     env.enableCheckpointing(5000) 
    25.   
  
  
  
  25.     env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE) 
    26.   
  
  
  
  26.     env.setStateBackend(new FsStateBackend("file:///D:/space/IJ/bigdata/src/main/scala/com/ddxygq/bigdata/flink/checkpoint/streaming/counter")) 
    27.   
  
  
  
  27.  
    28.   
  
  
  
  28.     // kafka 配置 
    29.   
  
  
  
  29.     val ZOOKEEPER_HOST = "hadoop01:2181,hadoop02:2181,hadoop03:2181" 
    30.   
  
  
  
  30.     val KAFKA_BROKERS = "hadoop01:9092,hadoop02:9092,hadoop03:9092" 
    31.   
  
  
  
  31.     val TRANSACTION_GROUP = "flink-count" 
    32.   
  
  
  
  32.     val TOPIC_NAME = "flink" 
    33.   
  
  
  
  33.     val kafkaProps = new Properties() 
    34.   
  
  
  
  34.     kafkaProps.setProperty("zookeeper.connect", ZOOKEEPER_HOST) 
    35.   
  
  
  
  35.     kafkaProps.setProperty("bootstrap.servers", KAFKA_BROKERS) 
    36.   
  
  
  
  36.     kafkaProps.setProperty("group.id", TRANSACTION_GROUP) 
    37.   
  
  
  
  37.  
    38.   
  
  
  
  38.     // watrmark 允许数据延迟时间 
    39.   
  
  
  
  39.     val MaxOutOfOrderness = 86400 * 1000L 
    40.   
  
  
  
  40.  
    41.   
  
  
  
  41.     val streamData: DataStream[(String, String, String)] = env.addSource( 
    42.   
  
  
  
  42.       new FlinkKafkaConsumer010[String](TOPIC_NAME, new SimpleStringSchema(), kafkaProps) 
    43.   
  
  
  
  43.     ).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[String](Time.milliseconds(MaxOutOfOrderness)) { 
    44.   
  
  
  
  44.       override def extractTimestamp(element: String): Long = { 
    45.   
  
  
  
  45.         val t = JSON.parseObject(element) 
    46.   
  
  
  
  46.         val time = JSON.parseObject(JSON.parseObject(t.getString("message")).getString("decrypted_data")).getString("time") 
    47.   
  
  
  
  47.         time.toLong 
    48.   
  
  
  
  48.       } 
    49.   
  
  
  
  49.     }).map(x => { 
    50.   
  
  
  
  50.       var date = "error" 
    51.   
  
  
  
  51.       var guid = "error" 
    52.   
  
  
  
  52.       var helperversion = "error" 
    53.   
  
  
  
  53.       try { 
    54.   
  
  
  
  54.         val messageJsonObject = JSON.parseObject(JSON.parseObject(x).getString("message")) 
    55.   
  
  
  
  55.         val datetime = messageJsonObject.getString("time") 
    56.   
  
  
  
  56.         date = datetime.split(" ")(0) 
    57.   
  
  
  
  57.         // hour = datetime.split(" ")(1).substring(0, 2) 
    58.   
  
  
  
  58.         val decrypted_data_string = messageJsonObject.getString("decrypted_data") 
    59.   
  
  
  
  59.         if (!"".equals(decrypted_data_string)) { 
    60.   
  
  
  
  60.           val decrypted_data = JSON.parseObject(decrypted_data_string) 
    61.   
  
  
  
  61.           guid = decrypted_data.getString("guid").trim 
    62.   
  
  
  
  62.           helperversion = decrypted_data.getString("helperversion") 
    63.   
  
  
  
  63.         } 
    64.   
  
  
  
  64.       } catch { 
    65.   
  
  
  
  65.         case e: Exception => { 
    66.   
  
  
  
  66.           println(e) 
    67.   
  
  
  
  67.         } 
    68.   
  
  
  
  68.       } 
    69.   
  
  
  
  69.       (date, helperversion, guid) 
    70.   
  
  
  
  70.     }) 
    71.   
  
  
  
  71.  
    72.   
  
  
  
  72.     val resultStream = streamData.keyBy(x => { 
    73.   
  
  
  
  73.       x._1 + x._2 
    74.   
  
  
  
  74.     }).timeWindow(Time.days(1)) 
    75.   
  
  
  
  75.       .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(1))) 
    76.   
  
  
  
  76.       .applyWith(("", new IntCounter(), Set.empty[Int], 0L, 0L))( 
    77.   
  
  
  
  77.         foldFunction = { 
    78.   
  
  
  
  78.           case ((_, cou, set, _, 0), item) => { 
    79.   
  
  
  
  79.             val date = item._1 
    80.   
  
  
  
  80.             val helperversion = item._2 
    81.   
  
  
  
  81.             val guid = item._3 
    82.   
  
  
  
  82.             cou.add(1) 
    83.   
  
  
  
  83.             (date + "_" + helperversion, cou, set + guid.hashCode, 0L, 0L) 
    84.   
  
  
  
  84.           } 
    85.   
  
  
  
  85.         } 
    86.   
  
  
  
  86.         , windowFunction = { 
    87.   
  
  
  
  87.           case (key, window, result) => { 
    88.   
  
  
  
  88.             result.map { 
    89.   
  
  
  
  89.               case (leixing, cou, set, _, _) => { 
    90.   
  
  
  
  90.                 (leixing, cou.getLocalValue, set.size, window.getStart, window.getEnd) 
    91.   
  
  
  
  91.               } 
    92.   
  
  
  
  92.             } 
    93.   
  
  
  
  93.           } 
    94.   
  
  
  
  94.         } 
    95.   
  
  
  
  95.       ).keyBy(0) 
    96.   
  
  
  
  96.       .flatMapWithState[(String, Int, Int, Long, Long),(Int, Int)]{ 
    97.   
  
  
  
  97.       case ((key, numpv, numuv, begin, end), curr) => 
    98.   
  
  
  
  98.  
    99.   
  
  
  
  99.         curr match { 
    100.   
  
  
  
  100.           case Some(numCurr) if numCurr == (numuv, numpv) => 
    101.   
  
  
  
  101.             (Seq.empty, Some((numuv, numpv))) //如果之前已经有相同的数据,则返回空结果 
    102.   
  
  
  
  102.           case _ => 
    103.   
  
  
  
  103.             (Seq((key, numpv, numuv, begin, end)), Some((numuv, numpv))) 
    104.   
  
  
  
  104.         } 
    105.   
  
  
  
  105.     } 
    106.   
  
  
  
  106.  
    107.   
  
  
  
  107.     // 最终结果 
    108.   
  
  
  
  108.     val resultedStream = resultStream.map(x => { 
    109.   
  
  
  
  109.       val keys = x._1.split("_") 
    110.   
  
  
  
  110.       val date = keys(0) 
    111.   
  
  
  
  111.       val helperversion = keys(1) 
    112.   
  
  
  
  112.       (date, helperversion, x._2, x._3) 
    113.   
  
  
  
  113.     }) 
    114.   
  
  
  
  114.  
    115.   
  
  
  
  115.     val resultPath = "D:\\space\\IJ\\bigdata\\src\\main\\scala\\com\\ddxygq\\bigdata\\flink\\streaming\\pvuv\\result" 
    116.   
  
  
  
  116.     resultedStream.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE) 
    117.   
  
  
  
  117.     env.execute("PvUvCount") 
    118.   
  
  
  
  118.  
    119.   
  
  
  
  119.   } 
    120.   
  
  
  
  120. } 
    121.

改进

其实这里还是需要set保存uv，难免对内存有压力，如果我们的集群不大，为了节省开支，我们可以使用外部媒介，如hbase的rowkey唯一性、redis的set数据结构，都是可以达到实时、快速去重的目的。

参考资料

https://flink.sojb.cn/dev/event_time.htm

lhttp://wuchong.me/blog/2016/05/20/flink-internals-streams-and-operations-on-streams

https://segmentfault.com/a/1190000006235690

网页名称：Flink实时计算Pv、Uv的几种方法
文章来源：http://www.shufengxianlan.com/qtweb/news2/377802.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

网站策划知识

同城分类信息

成都网站制作　　 　小程序开发　　 　成都添翼二手车　　 　香港空间腾讯云　　 　眉山主机托管　　 　成都网站排名　　 　西部信息服务器托管　　 　界面设计　　 　Android开发　　 　四川國際商會　　 　西部信息机房　　 　吉安网站建设　　 　南充电信机房托管　　 　达州电信机房　　 　手机网站　　 　成都水电改造　　 　成都企业网站维护　　 　租用服务器　　 　成都搬家公司　　 　古巴雪茄