Sparak-Streaming基于Offset消费Kafka数据

怼烎@ 2022-06-18 03:59 133阅读 0赞

[Sparak-Streaming基于Offset消费Kafka数据][Sparak-Streaming_Offset_Kafka]

原文http://blog.csdn.net/kwu\_ganymede/article/details/50930962

1、官方提供消费kafka的数据实例

**\[java\]**  [view plain][]   [copy][view plain]

[![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  // Hold a reference to the current offset ranges, so it can be used downstream  
2.  var offsetRanges = Array\[OffsetRange\]()  
3.    
4.  directKafkaStream.transform \{ rdd =>  
5.    offsetRanges = rdd.asInstanceOf\[HasOffsetRanges\].offsetRanges  
6.    rdd  
7.  \}.map \{  
8.            ...  
9.  \}.foreachRDD \{ rdd =>  
10.   **for** (o <- offsetRanges) \{  
11.     println(s"$\{o.topic\} $\{o.partition\} $\{o.fromOffset\} $\{o.untilOffset\}")  
12.   \}  
13.   ...  
14. \}

官网链接：

http://[Spark][].apache.org/docs/latest/streaming-kafka-integration.html

核心思想就是在，创建kafkastream同时获取偏移量，消费完的同时，执行更新操作。

2、自定义操作kafka数据的offset的manager

这里注意，需要把类，放在 org.apache.spark.streaming.kafka下面，与源码操作的包一致

**\[java\]**  [view plain][]   [copy][view plain]

[![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  **package** org.apache.spark.streaming.kafka  
2.    
3.  **import** kafka.common.TopicAndPartition  
4.  **import** kafka.message.MessageAndMetadata  
5.  **import** kafka.serializer.Decoder  
6.  **import** scala.reflect.ClassTag  
7.  **import** org.apache.spark.SparkException  
8.  **import** org.apache.spark.streaming.kafka.KafkaCluster.LeaderOffset  
9.  **import** org.apache.spark.rdd.RDD  
10. **import** org.apache.spark.streaming.dstream.InputDStream  
11. **import** org.apache.spark.streaming.StreamingContext  
12.   
13. **class** KafkaManager(val kafkaParams: Map\[String, String\]) **extends** Serializable \{  
14.   
15.   **private** val kc = **new** KafkaCluster(kafkaParams)  
16.   
17.   **private** val flag = 1150 \* 10000l  
18.   
19.   /\*\* 
20.    \* 创建数据流 
21.    \* @param ssc 
22.    \* @param kafkaParams 
23.    \* @param topics 
24.    \* @tparam K 
25.    \* @tparam V 
26.    \* @tparam KD 
27.    \* @tparam VD 
28.    \* @return 
29.    \*/  
30.   def createDirectStream\[K: ClassTag, V: ClassTag, KD <: Decoder\[K\]: ClassTag, VD <: Decoder\[V\]: ClassTag\](ssc: StreamingContext, kafkaParams: Map\[String, String\], topics: Set\[String\]): InputDStream\[(K, V)\] = \{  
31.     val groupId = kafkaParams.get("group.id").get  
32.     // 在zookeeper上读取offsets前先根据实际情况更新offsets    
33.     setOrUpdateOffsets(topics, groupId)  
34.   
35.     //从zookeeper上读取offset开始消费message    
36.     val messages = \{  
37.       val partitionsE = kc.getPartitions(topics)  
38.       **if** (partitionsE.isLeft)  
39.         **throw** **new** SparkException(s"get kafka partition failed: $\{partitionsE.left.get\}")  
40.       val partitions = partitionsE.right.get  
41.       val consumerOffsetsE = kc.getConsumerOffsets(groupId, partitions)  
42.       **if** (consumerOffsetsE.isLeft)  
43.         **throw** **new** SparkException(s"get kafka consumer offsets failed: $\{consumerOffsetsE.left.get\}")  
44.       val consumerOffsets = consumerOffsetsE.right.get  
45.       KafkaUtils.createDirectStream\[K, V, KD, VD, (K, V)\](  
46.         ssc, kafkaParams, consumerOffsets, (mmd: MessageAndMetadata\[K, V\]) => (mmd.key, mmd.message))  
47.     \}  
48.     messages  
49.   \}  
50.   
51.   /\*\* 
52.    \* 创建数据流前，根据实际消费情况更新消费offsets 
53.    \* @param topics 
54.    \* @param groupId 
55.    \*/  
56.   **private** def setOrUpdateOffsets(topics: Set\[String\], groupId: String): Unit = \{  
57.     topics.foreach(topic => \{  
58.       var hasConsumed = **true**  
59.       val partitionsE = kc.getPartitions(Set(topic))  
60.       **if** (partitionsE.isLeft)  
61.         **throw** **new** SparkException(s"get kafka partition failed: $\{partitionsE.left.get\}")  
62.       val partitions = partitionsE.right.get  
63.       val consumerOffsetsE = kc.getConsumerOffsets(groupId, partitions)  
64.       **if** (consumerOffsetsE.isLeft) hasConsumed = **false**  
65.       **if** (hasConsumed) \{ // 消费过  
66.         /\*\* 
67.          \* 如果streaming程序执行的时候出现kafka.common.OffsetOutOfRangeException， 
68.          \* 说明zk上保存的offsets已经过时了，即kafka的定时清理策略已经将包含该offsets的文件删除。 
69.          \* 针对这种情况，只要判断一下zk上的consumerOffsets和earliestLeaderOffsets的大小， 
70.          \* 如果consumerOffsets比earliestLeaderOffsets还小的话，说明consumerOffsets已过时, 
71.          \* 这时把consumerOffsets更新为earliestLeaderOffsets 
72.          \*/  
73.         val earliestLeaderOffsetsE = kc.getEarliestLeaderOffsets(partitions)  
74.         **if** (earliestLeaderOffsetsE.isLeft)  
75.           **throw** **new** SparkException(s"get earliest leader offsets failed: $\{earliestLeaderOffsetsE.left.get\}")  
76.         val earliestLeaderOffsets = earliestLeaderOffsetsE.right.get  
77.         val consumerOffsets = consumerOffsetsE.right.get  
78.   
79.         // 可能只是存在部分分区consumerOffsets过时，所以只更新过时分区的consumerOffsets为earliestLeaderOffsets  
80.         var offsets: Map\[TopicAndPartition, Long\] = Map()  
81.         consumerOffsets.foreach(\{  
82.           **case** (tp, n) =>  
83.             val earliestLeaderOffset = earliestLeaderOffsets(tp).offset  
84.             **if** (n < earliestLeaderOffset) \{  
85.               println("consumer group:" + groupId + ",topic:" + tp.topic + ",partition:" + tp.partition +  
86.                 " offsets已经过时，更新为" + earliestLeaderOffset)  
87.               offsets += (tp -> earliestLeaderOffset)  
88.             \}  
89.         \})  
90.         **if** (!offsets.isEmpty) \{  
91.           kc.setConsumerOffsets(groupId, offsets)  
92.         \}  
93.       \} **else** \{ // 没有消费过  
94.         val reset = kafkaParams.get("auto.offset.reset").map(\_.toLowerCase)  
95.         var leaderOffsets: Map\[TopicAndPartition, LeaderOffset\] = **null**  
96.         **if** (reset == Some("smallest")) \{  
97.           val leaderOffsetsE = kc.getEarliestLeaderOffsets(partitions)  
98.           **if** (leaderOffsetsE.isLeft)  
99.             **throw** **new** SparkException(s"get earliest leader offsets failed: $\{leaderOffsetsE.left.get\}")  
100.           leaderOffsets = leaderOffsetsE.right.get  
101.         \} **else** \{  
102.           val leaderOffsetsE = kc.getLatestLeaderOffsets(partitions)  
103.           **if** (leaderOffsetsE.isLeft)  
104.             **throw** **new** SparkException(s"get latest leader offsets failed: $\{leaderOffsetsE.left.get\}")  
105.           leaderOffsets = leaderOffsetsE.right.get  
106.         \}  
107.         val offsets = leaderOffsets.map \{  
108.           **case** (tp, offset) => (tp, offset.offset)  
109.         \}  
110.         kc.setConsumerOffsets(groupId, offsets)  
111.       \}  
112.     \})  
113.   \}  
114.   
115.   /\*\* 
116.    \* 更新zookeeper上的消费offsets 
117.    \* 把当前的消费记录，写入zk 
118.    \* 
119.    \* @param rdd 
120.    \*/  
121.   def updateZKOffsets(rdd: RDD\[(String, String)\]): Unit = \{  
122.     val groupId = kafkaParams.get("group.id").get  
123.     val offsetsList = rdd.asInstanceOf\[HasOffsetRanges\].offsetRanges  
124.   
125.     **for** (offsets <- offsetsList) \{  
126.       val topicAndPartition = TopicAndPartition(offsets.topic, offsets.partition)  
127.       val o = kc.setConsumerOffsets(groupId, Map((topicAndPartition, offsets.untilOffset)))  
128.       **if** (o.isLeft) \{  
129.         println(s"Error updating the offset to Kafka cluster: $\{o.left.get\}")  
130.       \}  
131.     \}  
132.   \}  
133.   
134.   /\*\* 
135.    \* 更新zookeeper上的消费offsets 
136.    \* 把当前的消费记录的offset往前推 
137.    \* 并写入zk 
138.    \* 
139.    \* @param rdd 
140.    \* @param day 
141.    \*/  
142.   def updateZKOffsetsFromoffsetRanges(offsetRanges: Array\[OffsetRange\], day: Double): Unit = \{  
143.     val groupId = kafkaParams.get("group.id").get  
144.   
145.     **for** (offsets <- offsetRanges) \{  
146.       val topicAndPartition = TopicAndPartition(offsets.topic, offsets.partition)  
147.   
148.       var offsetStreaming = 0l  
149.   
150.       println("offsets.untilOffset " + offsets.untilOffset)  
151.   
152.       /\*\* 
153.        \* 如果streaming挂掉，则从偏移量的前flag开始计算 
154.        \* 由于在streaming里的window函数中进行了去重处理 
155.        \* 因此不用担心数据重复的问题 
156.        \*/  
157.       **if** (offsets.untilOffset >= flag) \{  
158.         offsetStreaming = offsets.untilOffset - (flag \* day).toLong  
159.       \} **else** \{  
160.         offsetStreaming = 0  
161.       \}  
162.   
163.       println("offsetStreaming " + offsetStreaming)  
164.   
165.       val o = kc.setConsumerOffsets(groupId, Map((topicAndPartition, offsetStreaming)))  
166.       **if** (o.isLeft) \{  
167.         println(s"Error updating the offset to Kafka cluster: $\{o.left.get\}")  
168.       \}  
169.     \}  
170.   \}  
171. \}

提供两个方法

1） 从哪里结束消费，再从那里开始继续

2） 从结束消费向前移动偏移量的操作，然后再重新消费

3、使用案例

**\[java\]**  [view plain][]   [copy][view plain]

[![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  val conf = **new** SparkConf().setAppName("NewsTopNRealRankOffset")//.setMaster("local\[3\]");  
2.    
3.  conf.set("spark.streaming.blockInterval", "50ms");  
4.  conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")  
5.  conf.set("spark.storage.memoryFraction", "0.4") //executor分配给缓存的内存比例，默认为0.6即60%，剩下40%为task运行的内存，实际上40%是偏小的  
6.  conf.set("spark.locality.wait", "6000") //6000毫秒  
7.  conf.set("spark.streaming.kafka.maxRatePerPartition", "35000") // 限制每秒钟从topic的每个partition最多消费的消息条数  
8.    
9.  //shuffle优化  
10. conf.set("spark.shuffle.consolidateFiles", "true")  
11. conf.set("spark.reducer.maxSizeInFlight", "150m")  
12. conf.set("spark.shuffle.file.buffer", "128k")  
13. conf.set("spark.shuffle.io.maxRetries", "8")  
14. conf.set("spark.shuffle.io.retryWait", "6s")  
15. conf.set("spark.shuffle.memoryFraction", "0.3")  
16.   
17. val sc = **new** SparkContext(conf);  
18. val ssc = **new** StreamingContext(sc, Seconds(25));  
19. //缓存的数据  
20. ssc.remember(Minutes(60 \* 24 \* 2));  
21.   
22. val sqlContext = **new** HiveContext(sc);  
23.   
24. Logger.getLogger("org.apache.spark").setLevel(Level.WARN);  
25. Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.ERROR);  
26.   
27. //1.注册UDF  
28. val udf = UDFUtils();  
29. udf.registerUdf(sqlContext);  
30.   
31. //2.kafka数据处理  
32. val kafkaServiceOffset = KakfaServiceOffset();  
33.   
34. val brokers = "bdc46.hexun.com:9092,bdc53.hexun.com:9092,bdc54.hexun.com:9092";  
35. val kafkaParams = Map\[String, String\]("metadata.broker.list" -> brokers, "serializer.class" -> "kafka.serializer.StringEncoder",  
36.   "group.id" -> "tracklogPrdNewsTopN\_TrackLogT", "auto.offset.reset" -> "largest"); //largest smallest  
37.   
38. val topics = Set("TrackLogT");  
39.   
40. //通过自定义的KafkaManager获取kafka数据源  
41. val km = **new** KafkaManager(kafkaParams);  
42. val kafkaStream = km.createDirectStream\[String, String, StringDecoder, StringDecoder\](ssc, kafkaParams, topics);  
43. var offsetRanges = Array\[OffsetRange\]()  
44.   
45. //创建kafkastream同时获取偏移量  
46. val kafkaStreamOffset = kafkaStream.transform \{ rdd =>  
47.   offsetRanges = rdd.asInstanceOf\[HasOffsetRanges\].offsetRanges  
48.   rdd  
49. \}  
50.   
51. val urlClickLogPairsDStream = kafkaServiceOffset.kafkaDStreamForNewsOffset(ssc, kafkaStreamOffset);  
52.   
53. //3.缓存hive中的数据  
54. val cacheUtils = CacheUtils();  
55. cacheUtils.cacheHiveData(sqlContext);  
56.   
57. //4.缓存窗口函数数据处理  
58. val urlClickCountsDStream = urlClickLogPairsDStream.reduceByKeyAndWindow(  
59.   (v1: Int, v2: Int) => \{  
60.     v1 + v2  
61.   \},  
62.   Seconds(3600 \* 35),  
63.   Seconds(250));  
64.   
65. //5.处理业务逻辑  
66. urlClickCountsDStream.foreachRDD(urlClickCountsRDD => \{  
67.   val urlClickCountRowRDD = urlClickCountsRDD.map(tuple => \{  
68.     val datetime = tuple.\_1.split("\\001")(0) + " " + tuple.\_1.split("\\001")(1);  
69.     val cookieid = tuple.\_1.split("\\001")(2);  
70.     val url = tuple.\_1.split("\\001")(3);  
71.     val artId = tuple.\_1.split("\\001")(4);  
72.     val click\_count = tuple.\_2;  
73.     Row(datetime, cookieid, url, artId, click\_count);  
74.   \});  
75.   
76.   val structType = StructType(Array(  
77.     StructField("datetime", StringType, **true**),  
78.     StructField("cookieid", StringType, **true**),  
79.     StructField("url", StringType, **true**),  
80.     StructField("artId", StringType, **true**),  
81.     StructField("click\_count", IntegerType, **true**)));  
82.   
83.   val categoryProductCountDF = sqlContext.createDataFrame(urlClickCountRowRDD, structType);  
84.   
85.   categoryProductCountDF.registerTempTable("url\_click\_log\_day");  
86.   
87.   //消费完的同时，更新操作  
88.   km.updateZKOffsetsFromoffsetRanges(offsetRanges, 1);  
89.   
90.   cacuDayTopN(sqlContext);  
91.   cacuHourTopN(sqlContext);  
92. \});  
93.   
94. //6.启动streaming任务  
95. ssc.start();  
96. ssc.awaitTermination();

注意，设置"group.id" ，即为在kafka队列的偏移量标志在zookeeper中的名称。

[Sparak-Streaming_Offset_Kafka]: http://blog.csdn.net/kwu_ganymede/article/details/50930962
[view plain]: http://blog.csdn.net/kwu_ganymede/article/details/50930962#
[CODE]: https://code.csdn.net/assets/CODE_ico.png
[CODE_CODE]: https://code.csdn.net/snippets/1616081
[ico_fork.svg]: https://code.csdn.net/assets/ico_fork.svg
[ico_fork.svg 1]: https://code.csdn.net/snippets/1616081/fork
[Spark]: http://lib.csdn.net/base/spark