实时流计算、Spark Streaming、Kafka、Redis、Exactly-once、实时去重

ゝ一纸荒年。 2022-05-12 17:38 236阅读 0赞

在实时流式计算中，最重要的是在任何情况下，消息不重复、不丢失，即Exactly-once。本文以Kafka–>Spark Streaming–>Redis为例，一方面说明一下如何做到Exactly-once，另一方面说明一下我是如何计算实时去重指标的。

![spark streaming][]

## 1. 关于数据源 ##

数据源是文本格式的日志，由Nginx产生，存放于日志服务器上。在日志服务器上部署Flume Agent，使用TAILDIR Source和Kafka Sink，将日志采集到Kafka进行临时存储。日志格式如下：

2018-02-22T00:00:00+08:00|~|200|~|/test?**pcid**=DEIBAH&**siteid**=3

2018-02-22T00:00:00+08:00|~|200|~|/test?pcid=GLLIEG&siteid=3

2018-02-22T00:00:00+08:00|~|200|~|/test?pcid=HIJMEC&siteid=8

2018-02-22T00:00:00+08:00|~|200|~|/test?pcid=HMGBDE&siteid=3

2018-02-22T00:00:00+08:00|~|200|~|/test?pcid=HIJFLA&siteid=4

2018-02-22T00:00:01+08:00|~|200|~|/test?pcid=JCEBBC&siteid=9

2018-02-22T00:00:01+08:00|~|200|~|/test?pcid=KJLAKG&siteid=8

2018-02-22T00:00:01+08:00|~|200|~|/test?pcid=FHEIKI&siteid=3

2018-02-22T00:00:01+08:00|~|200|~|/test?pcid=IGIDLB&siteid=3

2018-02-22T00:00:01+08:00|~|200|~|/test?pcid=IIIJCD&siteid=5

日志是由测试程序模拟产生的，字段之间由|~|分隔。

## 2. 实时计算需求 ##

分天、分小时PV；

分天、分小时、分网站(siteid)PV；

分天 UV；

## 3. Spark Streaming消费Kafka数据 ##

http://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html

在Spark Streaming中消费Kafka数据，保证Exactly-once的核心有三点：

使用Direct方式连接Kafka；自己保存和维护Offset；更新Offset和计算在同一事务中完成；

后面的Spark Streaming程序（文章结尾），主要有以下步骤：

1.  启动后，先从Redis中获取上次保存的Offset，Redis中的key为”topic\_partition”，即每个分区维护一个Offset；
2.  使用获取到的Offset，创建DirectStream；
3.  在处理每批次的消息时，利用Redis的事务机制，确保在Redis中指标的计算和Offset的更新维护，在同一事务中完成。只有这两者同步，才能真正保证消息的Exactly-once。

1.  ./spark-submit \\
2.  \--class com.lxw1234.spark.TestSparkStreaming \\
3.  \--master local\[2\] \\
4.  \--conf spark.streaming.kafka.maxRatePerPartition=20000 \\
5.  \--jars /data1/home/dmp/lxw/realtime/commons-pool2-2.3.jar,\\
6.  /data1/home/dmp/lxw/realtime/jedis-2.9.0.jar,\\
7.  /data1/home/dmp/lxw/realtime/kafka-clients-0.11.0.1.jar,\\
8.  /data1/home/dmp/lxw/realtime/spark-streaming-kafka-0-10\_2.11-2.2.1.jar \\
9.  /data1/home/dmp/lxw/realtime/testsparkstreaming.jar \\
10. \--executor-memory 4G \\
11. \--num-executors 1

在启动Spark Streaming程序时候，有个参数最好指定：

spark.streaming.kafka.maxRatePerPartition=20000（每秒钟从topic的每个partition最多消费的消息条数）

如果程序第一次运行，或者因为某种原因暂停了很久重新启动时候，会积累很多消息，如果这些消息同时被消费，很有可能会因为内存不够而挂掉，因此，需要根据实际的数据量大小，以及批次的间隔时间来设置该参数，以限定批次的消息量。

如果该参数设置20000，而批次间隔时间未10秒，那么每个批次最多从Kafka中消费20万消息。

## 4. Redis中的数据模型 ##

*  分小时、分网站PV

普通K-V结构，计算时候使用incr命令递增，

Key为 “site\_pv\_网站ID\_小时”，

如：site\_pv\_9\_2018-02-21-00、site\_pv\_10\_2018-02-21-01

该数据模型用于计算分网站的按小时及按天PV。

*  分小时PV

普通K-V结构，计算时候使用incr命令递增，

Key为“pv\_小时”，如：pv\_2018-02-21-14、pv\_2018-02-22-03

该数据模型用于计算按小时及按天总PV。

*  分天UV

Set结构，计算时候使用sadd命令添加，

Key为”uv\_天”，如：uv\_2018-02-21、uv\_2018-02-20

该数据模型用户计算按天UV（获取时候使用SCARD命令获取Set元素个数）

注：这些Key对应的时间，均由实际消息中的第一个字段（时间）而定。

## 5. 故障恢复 ##

如果Spark Streaming程序因为停电、网络等意外情况终止而需要恢复，则直接重启即可；

如果因为其他原因需要重新计算某一时间段的消息，可以先删除Redis中对应时间段内的Key，然后从原始日志中截取该时间段内的消息，当做新消息添加至Kafka，由Spark Streaming程序重新消费并进行计算；

## 6. 附程序 ##

依赖jar包：

commons-pool2-2.3.jar

jedis-2.9.0.jar

kafka-clients-0.11.0.1.jar

spark-streaming-kafka-0-10\_2.11-2.2.1.jar

***InternalRedisClient （Redis链接池）***

1.  package com.lxw1234.spark
2.   
3.  import redis.clients.jedis.JedisPool
4.  import org.apache.commons.pool2.impl.GenericObjectPoolConfig
5.   
6.  /\*\*
7.  \* @author lxw1234
8.  \*/
9.  /\*\*
10. \* Internal Redis client for managing Redis connection \{@link Jedis\} based on \{@link RedisPool\}
11. \*/
12. object InternalRedisClient extends Serializable \{
13.  
14. @transient private var pool: JedisPool = null
15.  
16. def makePool(redisHost: String, redisPort: Int, redisTimeout: Int,
17. maxTotal: Int, maxIdle: Int, minIdle: Int): Unit = \{
18. makePool(redisHost, redisPort, redisTimeout, maxTotal, maxIdle, minIdle, true, false, 10000)
19. \}
20.  
21. def makePool(redisHost: String, redisPort: Int, redisTimeout: Int,
22. maxTotal: Int, maxIdle: Int, minIdle: Int, testOnBorrow: Boolean,
23. testOnReturn: Boolean, maxWaitMillis: Long): Unit = \{
24. if(pool == null) \{
25. val poolConfig = new GenericObjectPoolConfig()
26. poolConfig.setMaxTotal(maxTotal)
27. poolConfig.setMaxIdle(maxIdle)
28. poolConfig.setMinIdle(minIdle)
29. poolConfig.setTestOnBorrow(testOnBorrow)
30. poolConfig.setTestOnReturn(testOnReturn)
31. poolConfig.setMaxWaitMillis(maxWaitMillis)
32. pool = new JedisPool(poolConfig, redisHost, redisPort, redisTimeout)
33.  
34. val hook = new Thread\{
35. override def run = pool.destroy()
36. \}
37. sys.addShutdownHook(hook.run)
38. \}
39. \}
40.  
41. def getPool: JedisPool = \{
42. assert(pool != null)
43. pool
44. \}
45. \}

***TestSparkStreaming***

1.  package com.lxw1234.spark
2.   
3.  import org.apache.kafka.clients.consumer.ConsumerRecord
4.  import org.apache.kafka.common.TopicPartition
5.  import org.apache.kafka.common.serialization.StringDeserializer
6.  import org.apache.spark.SparkConf
7.  import org.apache.spark.rdd.RDD
8.  import org.apache.spark.streaming.Seconds
9.  import org.apache.spark.streaming.StreamingContext
10. import org.apache.spark.streaming.kafka010.ConsumerStrategies
11. import org.apache.spark.streaming.kafka010.HasOffsetRanges
12. import org.apache.spark.streaming.kafka010.KafkaUtils
13. import org.apache.spark.streaming.kafka010.LocationStrategies
14.  
15. import redis.clients.jedis.Pipeline
16.  
17.  
18. /\*\*
19. \* @author lxw1234
20. \* 获取topic最小的offset
21. \* ./kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list datadev1:9092 --topic lxw1234 --time -2
22. \*/
23. object TestSparkStreaming \{
24.  
25. def main(args : Array\[String\]) : Unit = \{
26. val brokers = "datadev1:9092"
27. val topic = "lxw1234"
28. val partition : Int = 0 //测试topic只有一个分区
29. val start\_offset : Long = 0l
30.  
31. //Kafka参数
32. val kafkaParams = Map\[String, Object\](
33. "bootstrap.servers" -> brokers,
34. "key.deserializer" -> classOf\[StringDeserializer\],
35. "value.deserializer" -> classOf\[StringDeserializer\],
36. "group.id" -> "exactly-once",
37. "enable.auto.commit" -> (false: java.lang.Boolean),
38. "auto.offset.reset" -> "none"
39. )
40.  
41. // Redis configurations
42. val maxTotal = 10
43. val maxIdle = 10
44. val minIdle = 1
45. val redisHost = "172.16.213.79"
46. val redisPort = 6379
47. val redisTimeout = 30000
48. //默认db，用户存放Offset和pv数据
49. val dbDefaultIndex = 8
50. InternalRedisClient.makePool(redisHost, redisPort, redisTimeout, maxTotal, maxIdle, minIdle)
51.  
52.  
53. val conf = new SparkConf().setAppName("TestSparkStreaming").setIfMissing("spark.master", "local\[2\]")
54. val ssc = new StreamingContext(conf, Seconds(10))
55.  
56. //从Redis获取上一次存的Offset
57. val jedis = InternalRedisClient.getPool.getResource
58. jedis.select(dbDefaultIndex)
59. val topic\_partition\_key = topic + "\_" + partition
60. var lastOffset = 0l
61. val lastSavedOffset = jedis.get(topic\_partition\_key)
62.  
63. if(null != lastSavedOffset) \{
64. try \{
65. lastOffset = lastSavedOffset.toLong
66. \} catch \{
67. case ex : Exception => println(ex.getMessage)
68. println("get lastSavedOffset error, lastSavedOffset from redis \[" + lastSavedOffset + "\] ")
69. System.exit(1)
70. \}
71. \}
72. InternalRedisClient.getPool.returnResource(jedis)
73.  
74. println("lastOffset from redis -> " + lastOffset)
75.  
76. //设置每个分区起始的Offset
77. val fromOffsets = Map\{new TopicPartition(topic, partition) -> lastOffset\}
78.  
79. //使用Direct API 创建Stream
80. val stream = KafkaUtils.createDirectStream\[String, String\](
81. ssc,
82. LocationStrategies.PreferConsistent,
83. ConsumerStrategies.Assign\[String, String\](fromOffsets.keys.toList, kafkaParams, fromOffsets)
84. )
85.  
86. //开始处理批次消息
87. stream.foreachRDD \{
88. rdd =>
89. val offsetRanges = rdd.asInstanceOf\[HasOffsetRanges\].offsetRanges
90.  
91. val result = processLogs(rdd)
92. println("=============== Total " + result.length + " events in this batch ..")
93.  
94. val jedis = InternalRedisClient.getPool.getResource
95. val p1 : Pipeline = jedis.pipelined();
96. p1.select(dbDefaultIndex)
97. p1.multi() //开启事务
98.  
99.  
100. //逐条处理消息
101. result.foreach \{
102. record =>
103. //增加小时总pv
104. val pv\_by\_hour\_key = "pv\_" + record.hour
105. p1.incr(pv\_by\_hour\_key)
106.  
107. //增加网站小时pv
108. val site\_pv\_by\_hour\_key = "site\_pv\_" + record.site\_id + "\_" + record.hour
109. p1.incr(site\_pv\_by\_hour\_key)
110.  
111. //使用set保存当天的uv
112. val uv\_by\_day\_key = "uv\_" + record.hour.substring(0, 10)
113. p1.sadd(uv\_by\_day\_key, record.user\_id)
114. \}
115.  
116. //更新Offset
117. offsetRanges.foreach \{ offsetRange =>
118. println("partition : " + offsetRange.partition + " fromOffset: " + offsetRange.fromOffset + " untilOffset: " + offsetRange.untilOffset)
119. val topic\_partition\_key = offsetRange.topic + "\_" + offsetRange.partition
120. p1.set(topic\_partition\_key, offsetRange.untilOffset + "")
121. \}
122.  
123. p1.exec();//提交事务
124. p1.sync();//关闭pipeline
125.  
126. InternalRedisClient.getPool.returnResource(jedis)
127.  
128. \}
129.  
130. case class MyRecord(hour: String, user\_id: String, site\_id: String)
131.  
132. def processLogs(messages: RDD\[ConsumerRecord\[String, String\]\]) : Array\[MyRecord\] = \{
133. messages.map(\_.value()).flatMap(parseLog).collect()
134. \}
135.  
136. //解析每条日志，生成MyRecord
137. def parseLog(line: String): Option\[MyRecord\] = \{
138. val ary : Array\[String\] = line.split("\\\\|~\\\\|", -1);
139. try \{
140. val hour = ary(0).substring(0, 13).replace("T", "-")
141. val uri = ary(2).split("\[=|&\]",-1)
142. val user\_id = uri(1)
143. val site\_id = uri(3)
144. return Some(MyRecord(hour,user\_id,site\_id))
145.  
146. \} catch \{
147. case ex : Exception => println(ex.getMessage)
148. \}
149.  
150. return None
151.  
152. \}
153.  
154.  
155.  
156.  
157. ssc.start()
158. ssc.awaitTermination()
159. \}
160.  
161.  
162. \}

![spark streaming][spark streaming 1]

![spark streaming][spark streaming 2]

[spark streaming]: /images/20220504/131fb80d133d4d918d5eaf2cac99cb37.png
[spark streaming 1]: /images/20220504/cb4ac80b28924592968bc9a67540f6a6.png
[spark streaming 2]: /images/20220504/0ae6dc5b931248caa121097bdf58bad1.png