Spark+Kafka的Direct方式将偏移量发送到Zookeeper实现

　Apache Spark 1.3.0引入了Direct API，利用Kafka的低层次API从Kafka集群中读取数据，而且在SparkStreaming系统里面维护偏移量相关的信息，而且经过这种方式去实现零数据丢失(zero data loss)相比使用基于Receiver的方法要高效。可是由于是Spark Streaming系统本身维护Kafka的读偏移量，而Spark Streaming系统并无将这个消费的偏移量发送到Zookeeper中，这将致使那些基于偏移量的Kafka集群监控软件（好比：Apache Kafka监控之Kafka Web Console、Apache Kafka监控之KafkaOffsetMonitor等）失效。本文就是基于为了解决这个问题，使得咱们编写的Spark Streaming程序可以在每次接收到数据以后自动地更新Zookeeper中Kafka的偏移量。

　　咱们从Spark的官方文档能够知道，维护Spark内部维护Kafka便宜了信息是存储在HasOffsetRanges类的offsetRanges中，咱们能够在Spark Streaming程序里面获取这些信息：

val offsetsList = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

这样咱们就能够获取因此分区消费信息，只须要遍历offsetsList，而后将这些信息发送到Zookeeper便可更新Kafka消费的偏移量。完整的代码片断以下：

val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)       messages.foreachRDD(rdd => {         val offsetsList = rdd.asInstanceOf[HasOffsetRanges].offsetRanges         val kc = new KafkaCluster(kafkaParams)         for (offsets < - offsetsList) {           val topicAndPartition = TopicAndPartition("iteblog", offsets.partition)           val o = kc.setConsumerOffsets(args(0), Map((topicAndPartition, offsets.untilOffset)))           if (o.isLeft) {             println(s"Error updating the offset to Kafka cluster: ${o.left.get}")           }         } })

　　KafkaCluster类用于创建和Kafka集群的连接相关的操做工具类，咱们能够对Kafka中Topic的每一个分区设置其相应的偏移量Map((topicAndPartition, offsets.untilOffset)),而后调用KafkaCluster类的setConsumerOffsets方法去更新Zookeeper里面的信息，这样咱们就能够更新Kafka的偏移量，最后咱们就能够经过KafkaOffsetMonitor之类软件去监控Kafka中相应Topic的消费信息，下图是KafkaOffsetMonitor的监控状况：

若是想及时了解Spark、Hadoop或者Hbase相关的文章，欢迎关注微信公共账号：iteblog_hadoop

　　从图中咱们能够看到KafkaOffsetMonitor监控软件已经能够监控到Kafka相关分区的消费状况，这对监控咱们整个Spark Streaming程序来很是重要，由于咱们能够任意时刻了解Spark读取速度。另外，KafkaCluster工具类的完整代码以下：

package org.apache.spark.streaming.kafka   import kafka.api.OffsetCommitRequest import kafka.common.{ErrorMapping, OffsetMetadataAndError, TopicAndPartition} import kafka.consumer.SimpleConsumer import org.apache.spark.SparkException import org.apache.spark.streaming.kafka.KafkaCluster.SimpleConsumerConfig   import scala.collection.mutable.ArrayBuffer import scala.util.Random import scala.util.control.NonFatal   /**  * User: 过往记忆  * Date: 2015-06-02  * Time: 下午23:46  * bolg: https://www.iteblog.com  * 本文地址：https://www.iteblog.com/archives/1381  * 过往记忆博客，专一于hadoop、hive、spark、shark、flume的技术博客，大量的干货  * 过往记忆博客微信公共账号：iteblog_hadoop  */   class KafkaCluster(val kafkaParams: Map[String, String]) extends Serializable {   type Err = ArrayBuffer[Throwable]     @transient private var _config: SimpleConsumerConfig = null     def config: SimpleConsumerConfig = this.synchronized {     if (_config == null) {       _config = SimpleConsumerConfig(kafkaParams)     }     _config   }     def setConsumerOffsets(groupId: String,                          offsets: Map[TopicAndPartition, Long]                           ): Either[Err, Map[TopicAndPartition, Short]] = {     setConsumerOffsetMetadata(groupId, offsets.map { kv =>       kv._1 -> OffsetMetadataAndError(kv._2)     })   }     def setConsumerOffsetMetadata(groupId: String,                                 metadata: Map[TopicAndPartition, OffsetMetadataAndError]                                  ): Either[Err, Map[TopicAndPartition, Short]] = {     var result = Map[TopicAndPartition, Short]()     val req = OffsetCommitRequest(groupId, metadata)     val errs = new Err     val topicAndPartitions = metadata.keySet     withBrokers(Random.shuffle(config.seedBrokers), errs) { consumer =>       val resp = consumer.commitOffsets(req)       val respMap = resp.requestInfo       val needed = topicAndPartitions.diff(result.keySet)       needed.foreach { tp: TopicAndPartition =>         respMap.get(tp).foreach { err: Short =>           if (err == ErrorMapping.NoError) {             result += tp -> err           } else {             errs.append(ErrorMapping.exceptionFor(err))           }         }       }       if (result.keys.size == topicAndPartitions.size) {         return Right(result)       }     }     val missing = topicAndPartitions.diff(result.keySet)     errs.append(new SparkException(s"Couldn't set offsets for ${missing}"))     Left(errs)   }     private def withBrokers(brokers: Iterable[(String, Int)], errs: Err)                          (fn: SimpleConsumer => Any): Unit = {     brokers.foreach { hp =>       var consumer: SimpleConsumer = null       try {         consumer = connect(hp._1, hp._2)         fn(consumer)       } catch {         case NonFatal(e) =>           errs.append(e)       } finally {         if (consumer != null) {           consumer.close()         }       }     }   }     def connect(host: String, port: Int): SimpleConsumer =     new SimpleConsumer(host, port, config.socketTimeoutMs,       config.socketReceiveBufferBytes, config.clientId) }