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

Spark SenLin 4年前 (2020-04-15) 407次浏览 已收录 0个评论

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

本文想记录和表达的东西挺多的,一时想不到什么好的标题,所以就用上面的关键字作为标题了。

在实时流式计算中,最重要的是在任何情况下,消息不重复、不丢失,即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/commonspool22.3.jar,\
  6. /data1/home/dmp/lxw/realtime/jedis2.9.0.jar,\
  7. /data1/home/dmp/lxw/realtime/kafkaclients0.11.0.1.jar,\
  8. /data1/home/dmp/lxw/realtime/sparkstreamingkafka010_2.112.2.1.jar \
  9. /data1/home/dmp/lxw/realtime/testsparkstreaming.jar \
  10. executormemory 4G \
  11. numexecutors 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. @transient private var pool: JedisPool = null
  14. def makePool(redisHost: String, redisPort: Int, redisTimeout: Int,
  15. maxTotal: Int, maxIdle: Int, minIdle: Int): Unit = {
  16. makePool(redisHost, redisPort, redisTimeout, maxTotal, maxIdle, minIdle, true, false, 10000)
  17. }
  18. def makePool(redisHost: String, redisPort: Int, redisTimeout: Int,
  19. maxTotal: Int, maxIdle: Int, minIdle: Int, testOnBorrow: Boolean,
  20. testOnReturn: Boolean, maxWaitMillis: Long): Unit = {
  21. if(pool == null) {
  22. val poolConfig = new GenericObjectPoolConfig()
  23. poolConfig.setMaxTotal(maxTotal)
  24. poolConfig.setMaxIdle(maxIdle)
  25. poolConfig.setMinIdle(minIdle)
  26. poolConfig.setTestOnBorrow(testOnBorrow)
  27. poolConfig.setTestOnReturn(testOnReturn)
  28. poolConfig.setMaxWaitMillis(maxWaitMillis)
  29. pool = new JedisPool(poolConfig, redisHost, redisPort, redisTimeout)
  30. val hook = new Thread{
  31. override def run = pool.destroy()
  32. }
  33. sys.addShutdownHook(hook.run)
  34. }
  35. }
  36. def getPool: JedisPool = {
  37. assert(pool != null)
  38. pool
  39. }
  40. }

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. def main(args : Array[String]) : Unit = {
  25. val brokers = “datadev1:9092”
  26. val topic = “lxw1234”
  27. val partition : Int = 0 //测试topic只有一个分区
  28. val start_offset : Long = 0l
  29. //Kafka参数
  30. val kafkaParams = Map[String, Object](
  31. “bootstrap.servers” -> brokers,
  32. “key.deserializer” -> classOf[StringDeserializer],
  33. “value.deserializer” -> classOf[StringDeserializer],
  34. “group.id” -> “exactly-once”,
  35. “enable.auto.commit” -> (false: java.lang.Boolean),
  36. “auto.offset.reset” -> “none”
  37. )
  38. // Redis configurations
  39. val maxTotal = 10
  40. val maxIdle = 10
  41. val minIdle = 1
  42. val redisHost = “172.16.213.79”
  43. val redisPort = 6379
  44. val redisTimeout = 30000
  45. //默认db,用户存放Offset和pv数据
  46. val dbDefaultIndex = 8
  47. InternalRedisClient.makePool(redisHost, redisPort, redisTimeout, maxTotal, maxIdle, minIdle)
  48. val conf = new SparkConf().setAppName(“TestSparkStreaming”).setIfMissing(“spark.master”, “local[2]”)
  49. val ssc = new StreamingContext(conf, Seconds(10))
  50. //从Redis获取上一次存的Offset
  51. val jedis = InternalRedisClient.getPool.getResource
  52. jedis.select(dbDefaultIndex)
  53. val topic_partition_key = topic + “_” + partition
  54. var lastOffset = 0l
  55. val lastSavedOffset = jedis.get(topic_partition_key)
  56. if(null != lastSavedOffset) {
  57. try {
  58. lastOffset = lastSavedOffset.toLong
  59. } catch {
  60. case ex : Exception => println(ex.getMessage)
  61. println(“get lastSavedOffset error, lastSavedOffset from redis [“ + lastSavedOffset + “] “)
  62. System.exit(1)
  63. }
  64. }
  65. InternalRedisClient.getPool.returnResource(jedis)
  66. println(“lastOffset from redis -> “ + lastOffset)
  67. //设置每个分区起始的Offset
  68. val fromOffsets = Map{new TopicPartition(topic, partition) -> lastOffset}
  69. //使用Direct API 创建Stream
  70. val stream = KafkaUtils.createDirectStream[String, String](
  71. ssc,
  72. LocationStrategies.PreferConsistent,
  73. ConsumerStrategies.Assign[String, String](fromOffsets.keys.toList, kafkaParams, fromOffsets)
  74. )
  75. //开始处理批次消息
  76. stream.foreachRDD {
  77. rdd =>
  78. val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
  79. val result = processLogs(rdd)
  80. println(“=============== Total “ + result.length + ” events in this batch ..”)
  81. val jedis = InternalRedisClient.getPool.getResource
  82. val p1 : Pipeline = jedis.pipelined();
  83. p1.select(dbDefaultIndex)
  84. p1.multi() //开启事务
  85. //逐条处理消息
  86. result.foreach {
  87. record =>
  88. //增加小时总pv
  89. val pv_by_hour_key = “pv_” + record.hour
  90. p1.incr(pv_by_hour_key)
  91. //增加网站小时pv
  92. val site_pv_by_hour_key = “site_pv_” + record.site_id + “_” + record.hour
  93. p1.incr(site_pv_by_hour_key)
  94. //使用set保存当天的uv
  95. val uv_by_day_key = “uv_” + record.hour.substring(0, 10)
  96. p1.sadd(uv_by_day_key, record.user_id)
  97. }
  98. //更新Offset
  99. offsetRanges.foreach { offsetRange =>
  100. println(“partition : “ + offsetRange.partition + ” fromOffset: “ + offsetRange.fromOffset + ” untilOffset: “ + offsetRange.untilOffset)
  101. val topic_partition_key = offsetRange.topic + “_” + offsetRange.partition
  102. p1.set(topic_partition_key, offsetRange.untilOffset + “”)
  103. }
  104. p1.exec();//提交事务
  105. p1.sync();//关闭pipeline
  106. InternalRedisClient.getPool.returnResource(jedis)
  107.  
  108. }
  109. case class MyRecord(hour: String, user_id: String, site_id: String)
  110. def processLogs(messages: RDD[ConsumerRecord[String, String]]) : Array[MyRecord] = {
  111. messages.map(_.value()).flatMap(parseLog).collect()
  112. }
  113. //解析每条日志,生成MyRecord
  114. def parseLog(line: String): Option[MyRecord] = {
  115. val ary : Array[String] = line.split(“\\|~\\|”, 1);
  116. try {
  117. val hour = ary(0).substring(0, 13).replace(“T”, “-“)
  118. val uri = ary(2).split(“[=|&]”,-1)
  119. val user_id = uri(1)
  120. val site_id = uri(3)
  121. return Some(MyRecord(hour,user_id,site_id))
  122. } catch {
  123. case ex : Exception => println(ex.getMessage)
  124. }
  125. return None
  126. }
  127. ssc.start()
  128. ssc.awaitTermination()
  129. }
  130.  
  131. }

 

spark streaming

 

spark streaming


top8488大数据 , 版权所有丨如未注明 , 均为原创丨本网站采用BY-NC-SA协议进行授权
转载请注明原文链接:实时流计算、Spark Streaming、Kafka、Redis、Exactly-once
喜欢 (0)
[]
分享 (0)

您必须 登录 才能发表评论!