Kafka spark directStream не может получить данные
Я использую spark directStream api для чтения данных из Kafka. Мой код выглядит следующим образом, Пожалуйста:
val sparkConf = new SparkConf().setAppName("testdirectStreaming")
val sc = new SparkContext(sparkConf)
val ssc = new StreamingContext(sc, Seconds(2))
val kafkaParams = Map[String, String](
"auto.offset.reset" -> "smallest",
"metadata.broker.list"->"10.0.0.11:9092",
"spark.streaming.kafka.maxRatePerPartition"->"100"
)
//I set all of the 3 partitions fromOffset are 0
var fromOffsets:Map[TopicAndPartition, Long] = Map(TopicAndPartition("mytopic",0) -> 0)
fromOffsets+=(TopicAndPartition("mytopic",1) -> 0)
fromOffsets+=(TopicAndPartition("mytopic",2) -> 0)
val kafkaData = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, MessageAndMetadata[String, String]](
ssc, kafkaParams, fromOffsets,(mmd: MessageAndMetadata[String, String]) => mmd)
var offsetRanges = Array[OffsetRange]()
kafkaData.transform { rdd =>
offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
rdd
}.map {
_.message()
}.foreachRDD { rdd =>
for (o <- offsetRanges) {
println(s"---${o.topic} ${o.partition} ${o.fromOffset} ${o.untilOffset}")
}
rdd.foreachPartition{ partitionOfRecords =>
partitionOfRecords.foreach { line =>
println("===============value:"+line)
}
}
}
Я уверен, что в кластере Кафки есть данные, но мой код не смог получить ни одного из них. Заранее спасибо.1 ответ:
Я нашел причину: старые сообщения в Кафке уже были удалены, так как срок хранения истек. Поэтому, когда я установил fromOffset равным 0, это вызвало исключение OutOfOffSet. Исключение вызвало то, что Искра сбросила смещение с последних. Поэтому я не мог получить никаких сообщений. Решение состоит в том, что мне нужно установить соответствующий fromOffset, чтобы избежать исключения.