请稍等 ...
×

采纳答案成功！

向帮助你的同学说点啥吧！感谢那些助人为乐的人

spark streaming运行较长时间后报错kafka.common.OffsetOutOfRangeException

运行几天之后就会出现这个问题，现在没有头绪，请老师指点一下可能出现的问题

完整日志放在最后面，分为driver和executor日志，我用的是standalone模式，一个worker。
这是我处理Kafka偏移的代码：

	val kafkaStream = KafkaHelperUtil.createDirectStream(ssc)
    var offsetRanges = Array[OffsetRange]()
    val data = kafkaStream.transform( rdd =>{
      offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      KafkaHelperUtil.updateZkOffsets(offsetRanges)
      rdd
    }).map(_._2)

KafkaHelperUtil是我自己封装的工具，为了避免内容太多暂时先不贴上来了。

Driver日志：

2019-03-07 09:47:00,916 [task-result-getter-2] [org.apache.spark.scheduler.TaskSetManager] [ERROR] - Task 0 in stage 4224213.0 failed 4 times; aborting job
2019-03-07 09:47:00,952 [JobScheduler] [org.apache.spark.streaming.scheduler.JobScheduler] [ERROR] - Error running job streaming job 1551739605000 ms.0
org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 4224213.0 failed 4 times, most recent failure: Lost task 0.3 in stage 4224213.0 (TID 32690, 0.0.0.0, executor 0): kafka.common.OffsetOutOfRangeException
	at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
	at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
	at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
	at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
	at java.lang.Class.newInstance(Class.java:442)
	at kafka.common.ErrorMapping$.exceptionFor(ErrorMapping.scala:86)
	at org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.handleFetchErr(KafkaRDD.scala:188)
	at org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.fetchBatch(KafkaRDD.scala:197)
	at org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.getNext(KafkaRDD.scala:212)
	at org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:73)
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
	at org.apache.spark.util.collection.ExternalSorter.insertAll(ExternalSorter.scala:191)
	at org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:63)
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96)
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
	at org.apache.spark.scheduler.Task.run(Task.scala:108)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:335)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

Driver stacktrace:
	at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1499)
	at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1487)
	at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1486)
	at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
	at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
	at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1486)
	at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:814)
	at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:814)
	at scala.Option.foreach(Option.scala:257)
	at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:814)
	at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1714)
	at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1669)
	at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1658)
	at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
	at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:630)
	at org.apache.spark.SparkContext.runJob(SparkContext.scala:2022)
	at org.apache.spark.SparkContext.runJob(SparkContext.scala:2043)
	at org.apache.spark.SparkContext.runJob(SparkContext.scala:2062)
	at org.apache.spark.SparkContext.runJob(SparkContext.scala:2087)
	at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1.apply(RDD.scala:926)
	at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1.apply(RDD.scala:924)
	at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
	at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
	at org.apache.spark.rdd.RDD.withScope(RDD.scala:362)
	at org.apache.spark.rdd.RDD.foreachPartition(RDD.scala:924)
	at com.bojianger.dyspark.storage.WindowOp$$anonfun$process$1.apply(WindowOp.scala:47)
	at com.bojianger.dyspark.storage.WindowOp$$anonfun$process$1.apply(WindowOp.scala:44)
	at org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3.apply(DStream.scala:628)
	at org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3.apply(DStream.scala:628)
	at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(ForEachDStream.scala:51)
	at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:51)
	at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:51)
	at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:416)
	at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply$mcV$sp(ForEachDStream.scala:50)
	at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:50)
	at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:50)
	at scala.util.Try$.apply(Try.scala:192)
	at org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)
	at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply$mcV$sp(JobScheduler.scala:257)
	at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:257)
	at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:257)
	at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
	at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
Caused by: kafka.common.OffsetOutOfRangeException
	at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
	at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
	at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
	at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
	at java.lang.Class.newInstance(Class.java:442)
	at kafka.common.ErrorMapping$.exceptionFor(ErrorMapping.scala:86)
	at org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.handleFetchErr(KafkaRDD.scala:188)
	at org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.fetchBatch(KafkaRDD.scala:197)
	at org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.getNext(KafkaRDD.scala:212)
	at org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:73)
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
	at org.apache.spark.util.collection.ExternalSorter.insertAll(ExternalSorter.scala:191)
	at org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:63)
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96)
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
	at org.apache.spark.scheduler.Task.run(Task.scala:108)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:335)
	... 3 more

Executor日志：

2019-03-07 09:47:00,650 [Executor task launch worker for task 32682] [org.apache.spark.executor.Executor] [ERROR] - Exception in task 0.0 in stage 4224213.0 (TID 32682)
kafka.common.OffsetOutOfRangeException
	at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
	at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
	at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
	at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
	at java.lang.Class.newInstance(Class.java:442)
	at kafka.common.ErrorMapping$.exceptionFor(ErrorMapping.scala:86)
	at org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.handleFetchErr(KafkaRDD.scala:188)
	at org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.fetchBatch(KafkaRDD.scala:197)
	at org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.getNext(KafkaRDD.scala:212)
	at org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:73)
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)
	at org.apache.spark.util.collection.ExternalSorter.insertAll(ExternalSorter.scala:191)
	at org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:63)
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96)
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
	at org.apache.spark.scheduler.Task.run(Task.scala:108)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:335)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
牛顿f5 2019-03-08 15:54:48
源自：10-10  -Direct方式整合之服务器环境联调
2447
收起

正在回答 回答被采纳积分+3

3回答

提问者 牛顿f5 2019-03-08 17:42:47

查看了一下官网代码，是不是在foreachRDD里面做保存offset操作更合理一点？跟直接在transform里面保存，除了保存时间的区别，还有其他区别吗？

// Hold a reference to the current offset ranges, so it can be used downstream

 var offsetRanges = Array.empty[OffsetRange]


 directKafkaStream.transform { rdd =>

   offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

   rdd

 }.map {

           ...

 }.foreachRDD { rdd =>

   for (o <- offsetRanges) {

     println(s"${o.topic} ${o.partition} ${o.fromOffset} ${o.untilOffset}")

   }

   ...

 }


0 回复 有任何疑惑可以回复我~
收起回答
取消 回复
Michael_PK 2019-03-08 16:51:42

这个要定位，建议采用如下的思路：你的把offset打印出户，等报错的时候拿报错的信息去kafka的offset里去判断，是否是真的offset的管理有问题

0 回复 有任何疑惑可以回复我~
收起回答
  • 提问者 牛顿f5 #1 
    应用在运行期间应该不会读取了我保存的offset吧？spark streaming拉取数据时所依赖的Kafka偏移量也是保存在zk里的吗？难道是我覆盖了Kafka原生的偏移量？请教Kafka原生的偏移量zk目录是：/consumers/0/offsets/topic_name/0类似这个格式吗
    回复 有任何疑惑可以回复我~ 2019-03-08 17:41:50
  • 提问者 牛顿f5 #2 
    有没有这种可能：Kafka缓冲区不够大，或者数据丢失，导致spark拉取丢失数据的时候越界呢？
    回复 有任何疑惑可以回复我~ 2019-03-08 17:45:41
  • Michael_PK 回复 提问者 牛顿f5 #3 
    每次批次处理完都会更新到zk
    回复 有任何疑惑可以回复我~ 2019-03-08 17:52:36
点开查看后面6条评论
取消 回复
提问者 牛顿f5 2019-03-08 15:58:06

是用zookeeper保存offset偏移量的

ZkUtils.updatePersistentPath(zkClient, newZkPath, o.fromOffset.toString)


0 回复 有任何疑惑可以回复我~
收起回答
取消 回复

相似问题

请教老师spark streaming能直接读取hbase或hive的数据，然后实时计算输出？
959 0 3
运行spark-shell的时候报错
848 0 4
PyCharm 本地运行找不到spark-submit
716 0 3
spark 写ES报错
989 0 10
flume kafka spark streaming
696 0 2

登录后可查看更多问答，登录/注册

问题已解决，确定采纳
还有疑问，暂不采纳
Spark Streaming实时流处理项目实战
  • 参与学习       1404    人
  • 解答问题       571    个

Flume+Kafka+Spark Streaming 构建通用实时流处理平台

了解课程

本课精华内容

问答

读英语的时候能不能不要propertie撕，topic撕

1.7k 89

cannot assign requested address

1.2k 14

kafka生产者消息发不出去

2.4k 13

kafka.serializer.StringDecoder死活导不进来

1.7k 12

spark streaming运行较长时间后报错kafka.common.OffsetOutOfRangeException

2.4k 12

查看更多本课问答
意见反馈 帮助中心 APP下载
官方微信