KafkaUtils 用于创建一个从Kafka Brokers 拉取数据的输入数据流。
之前有一个文章介绍了sparkstream创建kafka的数据流有两种方式，一种是Receiver 一种是Direct方式。我们先看下Direct方式，具体的区别可以参考我的另一篇文章https://www.jianshu.com/p/88862316c4db
代码深入：
KafkaUtils->DirectKafkaInputDStream
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] = {
val cleanedHandler = ssc.sc.clean(messageHandler)
new DirectKafkaInputDStream[K, V, KD, VD, R](
ssc, kafkaParams, fromOffsets, cleanedHandler)
}
入参如上所示 ssc，kafkaParams，topics，可以多个topic，storageLevel
DirectKafkaInputDStream KafkaRDD的stream，并且Kafka的每个topic的每个Partition 与RDD的partition一一对应。
spark.streaming.kafka.maxRatePerPartition 这个参数决定了每个partition每秒钟接收的最大的消息数量。并且这个Dstream并不负责提交offsets。因此你可以实现exactly-once 语义。
首先我们要看一下compute方法，也就是负责产生指定时间RDD的方法。这个方法我会在DStream里面提到。
override def compute(validTime: Time): Option[KafkaRDD[K, V, U, T, R]] = {
val untilOffsets = clamp(latestLeaderOffsets(maxRetries))//获取各个partition应该获取的offset 也就是当前的offset+maxRatePerPartition 和partiton最新的offset中取最小值。
val rdd = KafkaRDD[K, V, U, T, R](//根据当前的offset和最新的offset创建一个KafkaRdd
context.sparkContext, kafkaParams, currentOffsets, untilOffsets, messageHandler)
// Report the record number and metadata of this batch interval to InputInfoTracker.
val offsetRanges = currentOffsets.map { case (tp, fo) =>
val uo = untilOffsets(tp)
OffsetRange(tp.topic, tp.partition, fo, uo.offset)
}
val description = offsetRanges.filter { offsetRange =>
// Don't display empty ranges.
offsetRange.fromOffset != offsetRange.untilOffset
}.map { offsetRange =>
s"topic: ${offsetRange.topic}\tpartition: ${offsetRange.partition}\t" +
s"offsets: ${offsetRange.fromOffset} to ${offsetRange.untilOffset}"
}.mkString("\n")
// Copy offsetRanges to immutable.List to prevent from being modified by the user
val metadata = Map(
"offsets" -> offsetRanges.toList,
StreamInputInfo.METADATA_KEY_DESCRIPTION -> description)
val inputInfo = StreamInputInfo(id, rdd.count, metadata)
ssc.scheduler.inputInfoTracker.reportInfo(validTime, inputInfo)
//更新当前的offsets
currentOffsets = untilOffsets.map(kv => kv._1 -> kv._2.offset)
Some(rdd)
}
然后看下clamp方法 :获取各个partition应该获取的offset: 也就是当前的offset+maxRatePerPartition 和partiton最新的offset中取最小值。
protected def clamp(
leaderOffsets: Map[TopicAndPartition, LeaderOffset]): Map[TopicAndPartition, LeaderOffset] = {
maxMessagesPerPartition.map { mmp =>
leaderOffsets.map { case (tp, lo) =>
tp -> lo.copy(offset = Math.min(currentOffsets(tp) + mmp, lo.offset))
}
}.getOrElse(leaderOffsets)
}
再看获取最新offset的方法，入参是获取leaderoffset的重试参数。由于机器宕机等原因，某个partition的leader可能丢失，所以这个时候会有一个isr中的broker，成为该partition的leader。这样consumer就能够连接新的leader了。
@tailrec
protected final def latestLeaderOffsets(retries: Int): Map[TopicAndPartition, LeaderOffset] = {
val o = kc.getLatestLeaderOffsets(currentOffsets.keySet)//获取partition leader的offset
// Either.fold would confuse @tailrec, do it manually
if (o.isLeft) {//如果异常
val err = o.left.get.toString
if (retries <= 0) {//且重试次数为不大于0 的时候这抛出异常
throw new SparkException(err)
} else {//否则重试，并且线程等待 一定时间后，默认是200ms，这个可以通过修改refresh.leader.backoff.ms 参数修改
log.error(err)
Thread.sleep(kc.config.refreshLeaderBackoffMs)
latestLeaderOffsets(retries - 1)
}
} else {
o.right.get
}
}
这建议大家可以将重试次数设置成3，并且超时时间设置成3000。并且做好Job的运行状态检查，如果发现job异常退出的时候，可以自动重启Job。
KafkaRDD的创建参考KafkaRDD
https://www.jianshu.com/p/0b0767393d63
