Spark streaming kafka1.4.1中的低阶api createDirectStream使用总结(Scala实现)

青旅半醒 2022-06-18 02:18 80阅读 0赞

# S[park streaming kafka1.4.1中的低阶api createDirectStream使用总结][park streaming kafka1.4.1_api createDirectStream] #

# 原文：http://blog.csdn.net/ligt0610/article/details/47311771    #

由于目前每天需要从kafka中消费20亿条左右的消息，集群压力有点大，会导致job不同程度的异常退出。原来使用spark1.1.0版本中的createStream函数，但是在数据处理速度跟不上数据消费速度且job异常退出的情况下，可能造成大量的数据丢失。幸好，[Spark][]后续版本对这一情况有了很大的改进，1.2版本加入WAL特性，但是性能应该会受到一些影响（本人未[测试][Link 1]），1.3版本可以直接通过低阶API从kafka的topic消费消息，并且不再向zookeeper中更新consumer offsets，使得基于zookeeper的consumer offsets的监控工具都会失效。

官方只是非常简单的描述了可以用以下方法修改zookeeper中的consumer offsets（可以查看[http://spark.apache.org/docs/1.4.1/streaming-kafka-integration.html][http_spark.apache.org_docs_1.4.1_streaming-kafka-integration.html]）：

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

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.  \}

所以更新zookeeper中的consumer offsets还需要自己去实现，并且官方提供的两个createDirectStream重载并不能很好的满足我的需求，需要进一步封装。具体看以下KafkaManager类的代码：

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

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** org.apache.spark.SparkException  
7.  **import** org.apache.spark.rdd.RDD  
8.  **import** org.apache.spark.streaming.StreamingContext  
9.  **import** org.apache.spark.streaming.dstream.InputDStream  
10. **import** org.apache.spark.streaming.kafka.KafkaCluster.\{LeaderOffset\}  
11.   
12. **import** scala.reflect.ClassTag  
13.   
14. /\*\* 
15.  \* Created by knowpigxia on 15-8-5. 
16.  \*/  
17. **class** KafkaManager(val kafkaParams: Map\[String, String\]) **extends** Serializable \{  
18.   
19.   **private** val kc = **new** KafkaCluster(kafkaParams)  
20.   
21.   /\*\* 
22.    \* 创建数据流 
23.    \* @param ssc 
24.    \* @param kafkaParams 
25.    \* @param topics 
26.    \* @tparam K 
27.    \* @tparam V 
28.    \* @tparam KD 
29.    \* @tparam VD 
30.    \* @return 
31.    \*/  
32.   def createDirectStream\[K: ClassTag, V: ClassTag, KD <: Decoder\[K\]: ClassTag, VD <: Decoder\[V\]: ClassTag\](  
33.                                                                                                             ssc: StreamingContext, kafkaParams: Map\[String, String\], topics: Set\[String\]): InputDStream\[(K, V)\] =  \{  
34.     val groupId = kafkaParams.get("group.id").get  
35.     // 在zookeeper上读取offsets前先根据实际情况更新offsets  
36.     setOrUpdateOffsets(topics, groupId)  
37.   
38.     //从zookeeper上读取offset开始消费message  
39.     val messages = \{  
40.       val partitionsE = kc.getPartitions(topics)  
41.       **if** (partitionsE.isLeft)  
42.         **throw** **new** SparkException(s"get kafka partition failed: $\{partitionsE.left.get\}")  
43.       val partitions = partitionsE.right.get  
44.       val consumerOffsetsE = kc.getConsumerOffsets(groupId, partitions)  
45.       **if** (consumerOffsetsE.isLeft)  
46.         **throw** **new** SparkException(s"get kafka consumer offsets failed: $\{consumerOffsetsE.left.get\}")  
47.       val consumerOffsets = consumerOffsetsE.right.get  
48.       KafkaUtils.createDirectStream\[K, V, KD, VD, (K, V)\](  
49.         ssc, kafkaParams, consumerOffsets, (mmd: MessageAndMetadata\[K, V\]) => (mmd.key, mmd.message))  
50.     \}  
51.     messages  
52.   \}  
53.   
54.   /\*\* 
55.    \* 创建数据流前，根据实际消费情况更新消费offsets 
56.    \* @param topics 
57.    \* @param groupId 
58.    \*/  
59.   **private** def setOrUpdateOffsets(topics: Set\[String\], groupId: String): Unit = \{  
60.     topics.foreach(topic => \{  
61.       var hasConsumed = **true**  
62.       val partitionsE = kc.getPartitions(Set(topic))  
63.       **if** (partitionsE.isLeft)  
64.         **throw** **new** SparkException(s"get kafka partition failed: $\{partitionsE.left.get\}")  
65.       val partitions = partitionsE.right.get  
66.       val consumerOffsetsE = kc.getConsumerOffsets(groupId, partitions)  
67.       **if** (consumerOffsetsE.isLeft) hasConsumed = **false**  
68.       **if** (hasConsumed) \{ // 消费过  
69.         /\*\* 
70.          \* 如果streaming程序执行的时候出现kafka.common.OffsetOutOfRangeException， 
71.          \* 说明zk上保存的offsets已经过时了，即kafka的定时清理策略已经将包含该offsets的文件删除。 
72.          \* 针对这种情况，只要判断一下zk上的consumerOffsets和earliestLeaderOffsets的大小， 
73.          \* 如果consumerOffsets比earliestLeaderOffsets还小的话，说明consumerOffsets已过时, 
74.          \* 这时把consumerOffsets更新为earliestLeaderOffsets 
75.          \*/  
76.         val earliestLeaderOffsetsE = kc.getEarliestLeaderOffsets(partitions)  
77.         **if** (earliestLeaderOffsetsE.isLeft)  
78.           **throw** **new** SparkException(s"get earliest leader offsets failed: $\{earliestLeaderOffsetsE.left.get\}")  
79.         val earliestLeaderOffsets = earliestLeaderOffsetsE.right.get  
80.         val consumerOffsets = consumerOffsetsE.right.get  
81.   
82.         // 可能只是存在部分分区consumerOffsets过时，所以只更新过时分区的consumerOffsets为earliestLeaderOffsets  
83.         var offsets: Map\[TopicAndPartition, Long\] = Map()  
84.         consumerOffsets.foreach(\{ **case**(tp, n) =>  
85.           val earliestLeaderOffset = earliestLeaderOffsets(tp).offset  
86.           **if** (n < earliestLeaderOffset) \{  
87.             println("consumer group:" + groupId + ",topic:" + tp.topic + ",partition:" + tp.partition +  
88.               " offsets已经过时，更新为" + earliestLeaderOffset)  
89.             offsets += (tp -> earliestLeaderOffset)  
90.           \}  
91.         \})  
92.         **if** (!offsets.isEmpty) \{  
93.           kc.setConsumerOffsets(groupId, offsets)  
94.         \}  
95.       \} **else** \{ // 没有消费过  
96.       val reset = kafkaParams.get("auto.offset.reset").map(\_.toLowerCase)  
97.         var leaderOffsets: Map\[TopicAndPartition, LeaderOffset\] = **null**  
98.         **if** (reset == Some("smallest")) \{  
99.           val leaderOffsetsE = kc.getEarliestLeaderOffsets(partitions)  
100.           **if** (leaderOffsetsE.isLeft)  
101.             **throw** **new** SparkException(s"get earliest leader offsets failed: $\{leaderOffsetsE.left.get\}")  
102.           leaderOffsets = leaderOffsetsE.right.get  
103.         \} **else** \{  
104.           val leaderOffsetsE = kc.getLatestLeaderOffsets(partitions)  
105.           **if** (leaderOffsetsE.isLeft)  
106.             **throw** **new** SparkException(s"get latest leader offsets failed: $\{leaderOffsetsE.left.get\}")  
107.           leaderOffsets = leaderOffsetsE.right.get  
108.         \}  
109.         val offsets = leaderOffsets.map \{  
110.           **case** (tp, offset) => (tp, offset.offset)  
111.         \}  
112.         kc.setConsumerOffsets(groupId, offsets)  
113.       \}  
114.     \})  
115.   \}  
116.   
117.   /\*\* 
118.    \* 更新zookeeper上的消费offsets 
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. \}

接下来再给一个简单的例子：

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

1.  **import** kafka.serializer.StringDecoder  
2.  **import** org.apache.log4j.\{Level, Logger\}  
3.  **import** org.apache.spark.SparkConf  
4.  **import** org.apache.spark.rdd.RDD  
5.  **import** org.apache.spark.streaming.kafka.\_  
6.  **import** org.apache.spark.streaming.\{Seconds, StreamingContext\}  
7.    
8.  /\*\* 
9.   \* Created by knowpigxia on 15-8-4. 
10.  \*/  
11. object DirectKafkaWordCount \{  
12.   
13.   def dealLine(line: String): String = \{  
14.     val list = AnalysisUtil.dealString(line, ',', '"')// 把dealString函数当做split即可  
15.     list.get(0).substring(0, 10) + "-" + list.get(26)  
16.   \}  
17.   
18.   def processRdd(rdd: RDD\[(String, String)\]): Unit = \{  
19.     val lines = rdd.map(\_.\_2)  
20.     val words = lines.map(dealLine(\_))  
21.     val wordCounts = words.map(x => (x, 1L)).reduceByKey(\_ + \_)  
22.     wordCounts.foreach(println)  
23.   \}  
24.   
25.   def main(args: Array\[String\]) \{  
26.     **if** (args.length < 3) \{  
27.       System.err.println( s"""  
28.         |Usage: DirectKafkaWordCount <brokers> <topics> <groupid>  
29.         |  <brokers> is a list of one or more Kafka brokers  
30.         |  <topics> is a list of one or more kafka topics to consume from  
31.         |  <groupid> is a consume group  
32.         |  
33.         """.stripMargin)  
34.       System.exit(1)  
35.     \}  
36.   
37.     Logger.getLogger("org").setLevel(Level.WARN)  
38.   
39.     val Array(brokers, topics, groupId) = args  
40.   
41.     // Create context with 2 second batch interval  
42.     val sparkConf = **new** SparkConf().setAppName("DirectKafkaWordCount")  
43.     sparkConf.setMaster("local\[\*\]")  
44.     sparkConf.set("spark.streaming.kafka.maxRatePerPartition", "5")  
45.     sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")  
46.   
47.     val ssc = **new** StreamingContext(sparkConf, Seconds(2))  
48.   
49.     // Create direct kafka stream with brokers and topics  
50.     val topicsSet = topics.split(",").toSet  
51.     val kafkaParams = Map\[String, String\](  
52.       "metadata.broker.list" -> brokers,  
53.       "group.id" -> groupId,  
54.       "auto.offset.reset" -> "smallest"  
55.     )  
56.   
57.     val km = **new** KafkaManager(kafkaParams)  
58.   
59.     val messages = km.createDirectStream\[String, String, StringDecoder, StringDecoder\](  
60.       ssc, kafkaParams, topicsSet)  
61.   
62.     messages.foreachRDD(rdd => \{  
63.       **if** (!rdd.isEmpty()) \{  
64.         // 先处理消息  
65.         processRdd(rdd)  
66.         // 再更新offsets  
67.         km.updateZKOffsets(rdd)  
68.       \}  
69.     \})  
70.   
71.     ssc.start()  
72.     ssc.awaitTermination()  
73.   \}  
74. \}

[park streaming kafka1.4.1_api createDirectStream]: http://blog.csdn.net/ligt0610/article/details/47311771
[Spark]: http://lib.csdn.net/base/spark
[Link 1]: http://lib.csdn.net/base/softwaretest
[http_spark.apache.org_docs_1.4.1_streaming-kafka-integration.html]: http://spark.apache.org/docs/1.4.1/streaming-kafka-integration.html
[view plain]: http://blog.csdn.net/ligt0610/article/details/47311771#