标签：
Caused by: java.lang.AssertionError: assertion failed: Ran out of messages before reaching ending offset 264251742 for topic topic partition 1 start 264245135. This should not happen, and indicates that messages may have been lost
at scala.Predef$.assert(Predef.scala:179)
at org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.getNext(KafkaRDD.scala:165)
at org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:71)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
at scala.collection.Iterator$$anon$14.hasNext(Iterator.scala:388)
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
at org.apache.spark.util.collection.ExternalSorter.insertAll(ExternalSorter.scala:203)
at org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:73)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
at org.apache.spark.scheduler.Task.run(Task.scala:88)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
此处异常是由于kafka在读取指定offset日志时（此处是264245135到264251742），由于日志过大，导致日志的总大小超过 fetch.message.max.bytes的设定的值（默认为1024*1024），导致此错误。解决方法是，在kafka client的参数中加大fetch.message.max.bytes的值。
比如：
//kafka的配置文件
val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers,"fetch.message.max.bytes"->"10485760")
//工作流
val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)
Spark Streaming集成Kafak的问题之Ran out of messages
标签：
原文地址：http://www.cnblogs.com/luckuan/p/4968919.html
