Flink实时计算pv、uv的几种方法

今天药忘吃喽~ 2022-10-05 00:46 423阅读 0赞

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

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

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

![ab3edc6c3fd0b9dca691455618143184.png][]

## Flink数据流上的类型和操作 ##

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

![3d13050cb442c370d0e7b8db07c6e646.png][]

可以看出，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当中的。其它相关概念就不说了，会专门介绍，这里直接上代码吧。

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>com.ddxygq</groupId>
        <artifactId>bigdata</artifactId>
        <version>1.0-SNAPSHOT</version>
    
        <properties>
            <scala.version>2.11.8</scala.version>
            <flink.version>1.7.0</flink.version>
            <pkg.name>bigdata</pkg.name>
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-scala_2.11</artifactId>
                <version>{flink.version}</version>
      </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-scala_2.11</artifactId>
                <version>flink.version</version>
      </dependency>
      
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-java_2.11</artifactId>
                <version>{flink.version}</version>
            </dependency>
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-connector-kafka-0.10_2.11</artifactId>
                <version>flink.version</version>
      </dependency>
    
        <build>
            
            
            <finalName>basedir/src/test</testSourceDirectory>−−><finalName>{pkg.name}</finalName>
            <sourceDirectory>src/main/java</sourceDirectory>
            <resources>
                <resource>
                    <directory>src/main/resources</directory>
                    <includes>
                        <include>*.properties</include>
                        <include>*.xml</include>
                    </includes>
                    <filtering>false</filtering>
                </resource>
            </resources>
            <plugins>
                
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-surefire-plugin</artifactId>
                    <configuration>
                        <skip>true</skip>
                    </configuration>
                </plugin>
                
                <plugin>
                    <groupId>org.scala-tools</groupId>
                    <artifactId>maven-scala-plugin</artifactId>
                    <version>2.15.2</version>
                    <executions>
                        <execution>
                            <goals>
                                <goal>compile</goal>
                                <goal>testCompile</goal>
                            </goals>
                        </execution>
                    </executions>
                </plugin>
            </plugins>
        </build>
    </project>

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

package com.ddxygq.bigdata.flink.streaming.pvuv
    
    import java.util.Properties
    
    import com.alibaba.fastjson.JSON
    import org.apache.flink.runtime.state.filesystem.FsStateBackend
    import org.apache.flink.streaming.api.CheckpointingMode
    import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
    import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
    import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.streaming.api.windowing.triggers.ContinuousProcessingTimeTrigger
    import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010
    import org.apache.flink.streaming.util.serialization.SimpleStringSchema
    import org.apache.flink.streaming.api.scala.extensions._
    import org.apache.flink.api.scala._
    
    /**
      * @ Author: keguang
      * @ Date: 2019/3/18 17:34
      * @ version: v1.0.0
      * @ description: 
      */
    object PvUvCount {
      def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
    
        // 容错
        env.enableCheckpointing(5000)
        env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
        env.setStateBackend(new FsStateBackend("file:///D:/space/IJ/bigdata/src/main/scala/com/ddxygq/bigdata/flink/checkpoint/flink/tagApp"))
    
        // kafka 配置
        val ZOOKEEPER_HOST = "hadoop01:2181,hadoop02:2181,hadoop03:2181"
        val KAFKA_BROKERS = "hadoop01:9092,hadoop02:9092,hadoop03:9092"
        val TRANSACTION_GROUP = "flink-count"
        val TOPIC_NAME = "flink"
        val kafkaProps = new Properties()
        kafkaProps.setProperty("zookeeper.connect", ZOOKEEPER_HOST)
        kafkaProps.setProperty("bootstrap.servers", KAFKA_BROKERS)
        kafkaProps.setProperty("group.id", TRANSACTION_GROUP)
    
        // watrmark 允许数据延迟时间
        val MaxOutOfOrderness = 86400 * 1000L
        
        // 消费kafka数据
        val streamData: DataStream[(String, String, String)] = env.addSource(
          new FlinkKafkaConsumer010[String](TOPIC_NAME, new SimpleStringSchema(), kafkaProps)
        ).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[String](Time.milliseconds(MaxOutOfOrderness)) {
          override def extractTimestamp(element: String): Long = {
            val t = JSON.parseObject(element)
            val time = JSON.parseObject(JSON.parseObject(t.getString("message")).getString("decrypted_data")).getString("time")
            time.toLong
          }
        }).map(x => {
          var date = "error"
          var guid = "error"
          var helperversion = "error"
          try {
            val messageJsonObject = JSON.parseObject(JSON.parseObject(x).getString("message"))
            val datetime = messageJsonObject.getString("time")
            date = datetime.split(" ")(0)
            // hour = datetime.split(" ")(1).substring(0, 2)
            val decrypted_data_string = messageJsonObject.getString("decrypted_data")
            if (!"".equals(decrypted_data_string)) {
              val decrypted_data = JSON.parseObject(decrypted_data_string)
              guid = decrypted_data.getString("guid").trim
              helperversion = decrypted_data.getString("helperversion")
            }
          } catch {
            case e: Exception => {
              println(e)
            }
          }
          (date, helperversion, guid)
        })
        // 这上面是设置watermark并解析json部分
        // 聚合窗口中的数据，可以研究下applyWith这个方法和OnWindowedStream这个类
        val resultStream = streamData.keyBy(x => {
          x._1 + x._2
        }).timeWindow(Time.days(1))
          .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(1)))
          .applyWith(("", List.empty[Int], Set.empty[Int], 0L, 0L))(
            foldFunction = {
              case ((_, list, set, _, 0), item) => {
                val date = item._1
                val helperversion = item._2
                val guid = item._3
                (date + "_" + helperversion, guid.hashCode +: list, set + guid.hashCode, 0L, 0L)
              }
            }
            , windowFunction = {
              case (key, window, result) => {
                result.map {
                  case (leixing, list, set, _, _) => {
                    (leixing, list.size, set.size, window.getStart, window.getEnd)
                  }
                }
              }
            }
          ).keyBy(0)
          .flatMapWithState[(String, Int, Int, Long, Long),(Int, Int)]{
          case ((key, numpv, numuv, begin, end), curr) =>
    
            curr match {
              case Some(numCurr) if numCurr == (numuv, numpv) =>
                (Seq.empty, Some((numuv, numpv))) //如果之前已经有相同的数据,则返回空结果
              case _ =>
                (Seq((key, numpv, numuv, begin, end)), Some((numuv, numpv)))
            }
        }
    
        // 最终结果
        val resultedStream = resultStream.map(x => {
          val keys = x._1.split("_")
          val date = keys(0)
          val helperversion = keys(1)
          (date, helperversion, x._2, x._3)
        })
    
        resultedStream.print()
        env.execute("PvUvCount")
    
      }
    }

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

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

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

## 存在的问题 ##

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

## 改进版 ##

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

packagecom.ddxygq.bigdata.flink.streaming.pvuv
    
    import java.util.Properties
    
    import com.alibaba.fastjson.JSON
    import org.apache.flink.api.common.accumulators.IntCounter
    import org.apache.flink.runtime.state.filesystem.FsStateBackend
    import org.apache.flink.streaming.api.CheckpointingMode
    import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
    import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
    import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.streaming.api.windowing.triggers.ContinuousProcessingTimeTrigger
    import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010
    import org.apache.flink.streaming.util.serialization.SimpleStringSchema
    import org.apache.flink.streaming.api.scala.extensions._
    import org.apache.flink.api.scala._
    import org.apache.flink.core.fs.FileSystem
    
    object PvUv2 {
      def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
    
        // 容错
        env.enableCheckpointing(5000)
        env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
        env.setStateBackend(new FsStateBackend("file:///D:/space/IJ/bigdata/src/main/scala/com/ddxygq/bigdata/flink/checkpoint/streaming/counter"))
    
        // kafka 配置
        val ZOOKEEPER_HOST = "hadoop01:2181,hadoop02:2181,hadoop03:2181"
        val KAFKA_BROKERS = "hadoop01:9092,hadoop02:9092,hadoop03:9092"
        val TRANSACTION_GROUP = "flink-count"
        val TOPIC_NAME = "flink"
        val kafkaProps = new Properties()
        kafkaProps.setProperty("zookeeper.connect", ZOOKEEPER_HOST)
        kafkaProps.setProperty("bootstrap.servers", KAFKA_BROKERS)
        kafkaProps.setProperty("group.id", TRANSACTION_GROUP)
    
        // watrmark 允许数据延迟时间
        val MaxOutOfOrderness = 86400 * 1000L
    
        val streamData: DataStream[(String, String, String)] = env.addSource(
          new FlinkKafkaConsumer010[String](TOPIC_NAME, new SimpleStringSchema(), kafkaProps)
        ).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[String](Time.milliseconds(MaxOutOfOrderness)) {
          override def extractTimestamp(element: String): Long = {
            val t = JSON.parseObject(element)
            val time = JSON.parseObject(JSON.parseObject(t.getString("message")).getString("decrypted_data")).getString("time")
            time.toLong
          }
        }).map(x => {
          var date = "error"
          var guid = "error"
          var helperversion = "error"
          try {
            val messageJsonObject = JSON.parseObject(JSON.parseObject(x).getString("message"))
            val datetime = messageJsonObject.getString("time")
            date = datetime.split(" ")(0)
            // hour = datetime.split(" ")(1).substring(0, 2)
            val decrypted_data_string = messageJsonObject.getString("decrypted_data")
            if (!"".equals(decrypted_data_string)) {
              val decrypted_data = JSON.parseObject(decrypted_data_string)
              guid = decrypted_data.getString("guid").trim
              helperversion = decrypted_data.getString("helperversion")
            }
          } catch {
            case e: Exception => {
              println(e)
            }
          }
          (date, helperversion, guid)
        })
    
        val resultStream = streamData.keyBy(x => {
          x._1 + x._2
        }).timeWindow(Time.days(1))
          .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(1)))
          .applyWith(("", new IntCounter(), Set.empty[Int], 0L, 0L))(
            foldFunction = {
              case ((_, cou, set, _, 0), item) => {
                val date = item._1
                val helperversion = item._2
                val guid = item._3
                cou.add(1)
                (date + "_" + helperversion, cou, set + guid.hashCode, 0L, 0L)
              }
            }
            , windowFunction = {
              case (key, window, result) => {
                result.map {
                  case (leixing, cou, set, _, _) => {
                    (leixing, cou.getLocalValue, set.size, window.getStart, window.getEnd)
                  }
                }
              }
            }
          ).keyBy(0)
          .flatMapWithState[(String, Int, Int, Long, Long),(Int, Int)]{
          case ((key, numpv, numuv, begin, end), curr) =>
    
            curr match {
              case Some(numCurr) if numCurr == (numuv, numpv) =>
                (Seq.empty, Some((numuv, numpv))) //如果之前已经有相同的数据,则返回空结果
              case _ =>
                (Seq((key, numpv, numuv, begin, end)), Some((numuv, numpv)))
            }
        }
    
        // 最终结果
        val resultedStream = resultStream.map(x => {
          val keys = x._1.split("_")
          val date = keys(0)
          val helperversion = keys(1)
          (date, helperversion, x._2, x._3)
        })
    
        val resultPath = "D:\\space\\IJ\\bigdata\\src\\main\\scala\\com\\ddxygq\\bigdata\\flink\\streaming\\pvuv\\result"
        resultedStream.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE)
        env.execute("PvUvCount")
    
      }
    }

改进

其实这里还是需要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

**猜你喜欢**

[SparkStreaming实时计算pv和uv][SparkStreaming_pv_uv 1]

[Flink状态管理与状态一致性（长文）][Flink]

[Flink实时计算topN热榜][Flink_topN]

[数仓建模分层理论][Link 1]

[数仓建模方法论][Link 2]

[学习建议，大数据组件重点学习这几个][Link 3]

[SparkStreaming_pv_uv]: https://mp.weixin.qq.com/s?__biz=MzU5NzAwNDQ4NQ%3D%3D&idx=1&mid=2247488057&scene=21&sn=ae3f524d6b9e35d13a0c8959205cd6b0#wechat_redirect
[ab3edc6c3fd0b9dca691455618143184.png]: /images/20221005/66881d0242264db6a663e526341aea99.png
[3d13050cb442c370d0e7b8db07c6e646.png]: /images/20221005/a2acd09fd1a4493e8908830938939dc0.png
[SparkStreaming_pv_uv 1]: http://mp.weixin.qq.com/s?__biz=MzU5NzAwNDQ4NQ%3D%3D&chksm=fe5b4b93c92cc28500455483955b0dadcbd47fdd0a28b1e717f10f248772a7b030a06fc52878&idx=1&mid=2247488057&scene=21&sn=ae3f524d6b9e35d13a0c8959205cd6b0#wechat_redirect
[Flink]: http://mp.weixin.qq.com/s?__biz=MzU5NzAwNDQ4NQ%3D%3D&chksm=fe5b48a6c92cc1b0e54cf60e5daa028d24bc8776f25259925926f9cf33bb051e39249d0273ca&idx=1&mid=2247487756&scene=21&sn=f11ce03880a6e2979d4613c42f7f2188#wechat_redirect
[Flink_topN]: http://mp.weixin.qq.com/s?__biz=MzU5NzAwNDQ4NQ%3D%3D&chksm=fe5b5696c92cdf8066cd82c9d3e95d57fc60791f00153320a1d0d822aa06c9b51b59b3ac1eb6&idx=1&mid=2247487292&scene=21&sn=b6c487b0854fb5dc8a2859f917f6a339#wechat_redirect
[Link 1]: http://mp.weixin.qq.com/s?__biz=MzU5NzAwNDQ4NQ%3D%3D&chksm=fe5b525cc92cdb4a6842f5f48348b35bd10d265528c3b4861c1e89b5e09b84ab986282aa4a55&idx=1&mid=2247486454&scene=21&sn=c38a6f1a4020226adb0103bad03a82d1#wechat_redirect
[Link 2]: http://mp.weixin.qq.com/s?__biz=MzU5NzAwNDQ4NQ%3D%3D&chksm=fe5b5244c92cdb522f25d5a80e54fb5578837e78e93ad9a609e5cf8ec405d7e7ab0f0076eed1&idx=1&mid=2247486446&scene=21&sn=579828fcfd32a6a73d35f219df227ad9#wechat_redirect
[Link 3]: http://mp.weixin.qq.com/s?__biz=MzU5NzAwNDQ4NQ%3D%3D&chksm=fe5b4876c92cc1604b9be29b50a542a649cf01f422c8e7279ddeb80baac69d5620f0a6f1d427&idx=1&mid=2247487964&scene=21&sn=cc2a5da694e6077a235f9b30096d0cf1#wechat_redirect