KafkaUtils.createDirectStream | 程序员的自我修养 | 关注Java、大数据、机器学习

官网上对这个新接口的介绍很多，大致就是不与zookeeper交互，直接去kafka中读取数据，自己维护offset，于是速度比KafkaUtils.createStream要快上很多。但有利就有弊：无法进行offset的监控。

项目中需要尝试使用这个接口，同时还要进行offset的监控，于是只能按照官网所说的，自己将offset写入zookeeper。

方法1

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)] {...}

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)] {...}

这个方法只有3个参数，使用起来最为方便，但是每次启动的时候默认从Latest offset开始读取，或者设置参数auto.offset.reset="smallest"后将会从Earliest offset开始读取。

显然这2种读取位置都不适合生产环境。

方法2

def createDirectStream[
    K: ClassTag,
    V: ClassTag,
    KD <: Decoder[K]: ClassTag,
    VD <: Decoder[V]: ClassTag,
    R: ClassTag] (
      ssc: StreamingContext,
      kafkaParams: Map[String, String],
      fromOffsets: Map[TopicAndPartition, Long],
      messageHandler: MessageAndMetadata[K, V] => R
  ): InputDStream[R] = {...}

def createDirectStream[

K: ClassTag,

V: ClassTag,

KD <: Decoder[K]: ClassTag,

VD <: Decoder[V]: ClassTag,

R: ClassTag] (

ssc: StreamingContext,

kafkaParams: Map[String, String],

fromOffsets: Map[TopicAndPartition, Long],

messageHandler: MessageAndMetadata[K, V] => R

): InputDStream[R] = {...}

这个方法可以在启动的时候可以设置offset，但参数设置起来复杂很多，首先是fromOffsets: Map[TopicAndPartition, Long]的设置，参考下方代码。

val topic2Partitions = ZkUtils.getPartitionsForTopics(zkClient, Config.kafkaConfig.topic)
var fromOffsets: Map[TopicAndPartition, Long] = Map()

topic2Partitions.foreach(topic2Partitions => {
  val topic:String = topic2Partitions._1
  val partitions:Seq[Int] = topic2Partitions._2
  val topicDirs = new ZKGroupTopicDirs(Config.kafkaConfig.kafkaGroupId, topic)

  partitions.foreach(partition => {
    val zkPath = s"${topicDirs.consumerOffsetDir}/$partition"
    ZkUtils.makeSurePersistentPathExists(zkClient, zkPath)
    val untilOffset = zkClient.readData[String](zkPath)

    val tp = TopicAndPartition(topic, partition)
    val offset = try {
      if (untilOffset == null || untilOffset.trim == "")
        getMaxOffset(tp)
      else
        untilOffset.toLong
    } catch {
      case e: Exception => getMaxOffset(tp)
    }
    fromOffsets += (tp -> offset)
    logger.info(s"Offset init: set offset of $topic/$partition as $offset")

  })
})

val topic2Partitions = ZkUtils.getPartitionsForTopics(zkClient, Config.kafkaConfig.topic)

var fromOffsets: Map[TopicAndPartition, Long] = Map()

topic2Partitions.foreach(topic2Partitions => {

val topic:String = topic2Partitions._1

val partitions:Seq[Int] = topic2Partitions._2

val topicDirs = new ZKGroupTopicDirs(Config.kafkaConfig.kafkaGroupId, topic)

partitions.foreach(partition => {

val zkPath = s"${topicDirs.consumerOffsetDir}/$partition"

ZkUtils.makeSurePersistentPathExists(zkClient, zkPath)

val untilOffset = zkClient.readData[String](zkPath)

val tp = TopicAndPartition(topic, partition)

val offset = try {

if (untilOffset == null || untilOffset.trim == "")

getMaxOffset(tp)

else

untilOffset.toLong

} catch {

case e: Exception => getMaxOffset(tp)

}

fromOffsets += (tp -> offset)

logger.info(s"Offset init: set offset of $topic/$partition as $offset")

})

其中getMaxOffset方法是用来获取最大的offset。当第一次启动spark任务或者zookeeper上的数据被删除或设置出错时，将选取最大的offset开始消费。代码如下：

private def getMaxOffset(tp:TopicAndPartition):Long = {
  val request = OffsetRequest(immutable.Map(tp -> PartitionOffsetRequestInfo(OffsetRequest.LatestTime, 1)))

  ZkUtils.getLeaderForPartition(zkClient, tp.topic, tp.partition) match {
    case Some(brokerId) => {
      ZkUtils.readDataMaybeNull(zkClient, ZkUtils.BrokerIdsPath + "/" + brokerId)._1 match {
        case Some(brokerInfoString) => {
          Json.parseFull(brokerInfoString) match {
            case Some(m) =>
              val brokerInfo = m.asInstanceOf[Map[String, Any]]
              val host = brokerInfo.get("host").get.asInstanceOf[String]
              val port = brokerInfo.get("port").get.asInstanceOf[Int]
              new SimpleConsumer(host, port, 10000, 100000, "getMaxOffset")
                .getOffsetsBefore(request)
                .partitionErrorAndOffsets(tp)
                .offsets
                .head
            case None =>
              throw new BrokerNotAvailableException("Broker id %d does not exist".format(brokerId))
          }
        }
        case None =>
          throw new BrokerNotAvailableException("Broker id %d does not exist".format(brokerId))
      }
    }
    case None =>
      throw new Exception("No broker for partition %s - %s".format(tp.topic, tp.partition))
  }
}

private def getMaxOffset(tp:TopicAndPartition):Long = {

val request = OffsetRequest(immutable.Map(tp -> PartitionOffsetRequestInfo(OffsetRequest.LatestTime, 1)))

ZkUtils.getLeaderForPartition(zkClient, tp.topic, tp.partition) match {

case Some(brokerId) => {

ZkUtils.readDataMaybeNull(zkClient, ZkUtils.BrokerIdsPath + "/" + brokerId)._1 match {

case Some(brokerInfoString) => {

Json.parseFull(brokerInfoString) match {

case Some(m) =>

val brokerInfo = m.asInstanceOf[Map[String, Any]]

val host = brokerInfo.get("host").get.asInstanceOf[String]

val port = brokerInfo.get("port").get.asInstanceOf[Int]

new SimpleConsumer(host, port, 10000, 100000, "getMaxOffset")

.getOffsetsBefore(request)

.partitionErrorAndOffsets(tp)

.offsets

.head

case None =>

throw new BrokerNotAvailableException("Broker id %d does not exist".format(brokerId))

}

case None =>

throw new BrokerNotAvailableException("Broker id %d does not exist".format(brokerId))

}

case None =>

throw new Exception("No broker for partition %s - %s".format(tp.topic, tp.partition))

}

然后是参数messageHandler的设置，为了后续处理中能获取到topic，这里形成(topic, message)的tuple：

val messageHandler = (mmd: MessageAndMetadata[String, String]) => (mmd.topic, mmd.message())

1	val messageHandler = (mmd: MessageAndMetadata[String, String]) => (mmd.topic, mmd.message())

接着将从获取rdd的offset并写入到zookeeper中：

var offsetRanges = Array[OffsetRange]()
messages.transform { rdd =>
  offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
  rdd
}.foreachRDD(rdd => {
  rdd.foreachPartition(HBasePuter.batchSave)
  offsetRanges.foreach(o => {
    val topicDirs = new ZKGroupTopicDirs(Config.kafkaConfig.kafkaGroupId, o.topic)
    val zkPath = s"${topicDirs.consumerOffsetDir}/${o.partition}"
    ZkUtils.updatePersistentPath(zkClient, zkPath, o.untilOffset.toString)
    logger.info(s"Offset update: set offset of ${o.topic}/${o.partition} as ${o.untilOffset.toString}")
  })
})

var offsetRanges = Array[OffsetRange]()

messages.transform { rdd =>

offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

rdd

}.foreachRDD(rdd => {

rdd.foreachPartition(HBasePuter.batchSave)

offsetRanges.foreach(o => {

val topicDirs = new ZKGroupTopicDirs(Config.kafkaConfig.kafkaGroupId, o.topic)

val zkPath = s"${topicDirs.consumerOffsetDir}/${o.partition}"

ZkUtils.updatePersistentPath(zkClient, zkPath, o.untilOffset.toString)

logger.info(s"Offset update: set offset of ${o.topic}/${o.partition} as ${o.untilOffset.toString}")

})

最后附上batchSave的示例：

def batchSave(iter:Iterator[(String,String)]):Unit = {
  iter.foreach(item => {
    val topic = item._1
    val message = item._2
    ...
  })
}

def batchSave(iter:Iterator[(String,String)]):Unit = {

iter.foreach(item => {

val topic = item._1

val message = item._2

...

})

}

（转载本站文章请注明作者和出处程序员的自我修养 – SelfUp.cn ，请勿用于任何商业用途）

张瑞昌: 有很多，比较常见的是Jacob迭代法，一次迭代O(n^3)，迭代次数不清楚。如果是手动算的话按照定义求就可以了
Anonymous:
lc277: 你好我想问下一般删除节点要多久，要删除的datanode大概用了 1t，解除授权已经30多小时还没完成，请问是出现什么问题了吗麻烦告诉下谢谢 qq1844554123
Anonymous: 你好我想问下一般删除节点要多久，要删除的datanode大概用了 1t，解除授权已经30多小时还没完成，请问是出现什么问题了吗
Anonymous:
李雪璇: 想要完整代码，可以帮忙发给我吗
Anonymous: 请问一下,那个 user的推荐结果楼主查看了么? 为什么输入数据最高是五分,输出结果都是7分8分啥的?怎么设置输出的分数的最大值?
Anonymous: 那个 user的推荐结果楼主查看了么? 为什么输入数据最高是五分,输出结果都是7分8分啥的?
Anonymous: stopGracefullyOnShutdown在yarn- client模式下我测试的无效，你的呢
Anonymous: 另外,import的lib包能否发个列表.
Anonymous: 部分程序已经无法使用, 另外你的import代码也应该放上来哈
wzq: 赞
增达网: 受教了！呵呵！
Anonymous:
27: dsa会报错，rsa不会
Anonymous: 看错了忽略我
Anonymous: UserSideCF这个类在哪里
晴子: 在主节点初始化CM5数据库的时候报错误：Verifying that we can write to /opt/cm-5.9.0/etc/cloudera-scm -server log4j:ERROR Could not...
zhangnew: 就4题
linxh: “ 但要是遇到预先并不知道数组的长度而又需要获取正确的（或者称之为原始的）split长度时，该如何处理呢。。？ ” 印象中可以split函数参数传-1？

8条评论

匿名说道：

2015年12月5日下午10:05

回复
匿名说道：

2016年3月15日下午4:37

回复
匿名说道：

2016年6月30日下午4:21

网站做的挺漂亮的

回复
匿名说道：

2016年12月13日下午5:07

赞

回复
匿名说道：

2016年12月14日上午11:32

很不错，试问有java版的吗？

回复
匿名说道：

2016年12月28日上午9:56

ZkUtils.getPartitionsForTopics(zkClient, Config.kafkaConfig.topic) 你确定 kafka 里面有这个类？个人在kafka 最新稳定版几个版本中都没有找到对应的接口请指教

回复
- 匿名说道：
  
  2016年12月28日上午10:16
  
  ZkUtils.getPartitionsForTopics(zkClient, Config.kafkaConfig.topic)
  那个方法是在 spark-streaming_2.10 中 kafka 中没有对应的方法
  个人建议把对应的 import 带上
  
  回复
匿名说道：

2016年12月28日下午2:56

Config.kafkaConfig.kafkaGroupId 这个是指自己配置的group id 还是从 import org.apache.kafka.common.config.Config 这个类
如果是import org.apache.kafka.common.config.Config 这里面没有对应的API

回复

发表评论

点击这里取消回复。