sparkstreaming集成kafka，使用window时出现错误

  val ssc = new StreamingContext(sc, Seconds(5))

  val stream = KafkaUtils.createDirectStream(ssc,

    PreferConsistent,

    Assign[String, String](fromOffsets.keys.toList, kafkaParams, fromOffsets))

   

  val data = stream.map(_.value())

  val word = data.map(word => (word,1))

  val result = data.reduceByKeyAndWindow({ (x, y) => x + y }, { (x, y) => x - y }, Minutes(2),Minutes(1))

  word.print()

  result.print()

Exception in thread "main" org.apache.spark.SparkException: Job aborted due to stage failure: Task 1 in stage 34.0 failed 1 times, most recent failure: Lost task 1.0 in stage 34.0 (TID 126, localhost): java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access

    at org.apache.kafka.clients.consumer.KafkaConsumer.acquire(KafkaConsumer.java:1430)

    at org.apache.kafka.clients.consumer.KafkaConsumer.seek(KafkaConsumer.java:1131)

    at org.apache.spark.streaming.kafka010.CachedKafkaConsumer.seek(CachedKafkaConsumer.scala:95)

    at org.apache.spark.streaming.kafka010.CachedKafkaConsumer.get(CachedKafkaConsumer.scala:69)

    at org.apache.spark.streaming.kafka010.KafkaRDD$KafkaRDDIterator.next(KafkaRDD.scala:227)

    at org.apache.spark.streaming.kafka010.KafkaRDD$KafkaRDDIterator.next(KafkaRDD.scala:193)

    at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)

    at scala.collection.Iterator$$anon$12.nextCur(Iterator.scala:434)

    at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:440)

    at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)

    at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:439)

    at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)

    at org.apache.spark.util.collection.ExternalSorter.insertAll(ExternalSorter.scala:192)

    at org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:63)

    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:79)

    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:47)

    at org.apache.spark.scheduler.Task.run(Task.scala:85)

    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)

    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)

    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)

    at java.lang.Thread.run(Thread.java:745)



Driver stacktrace:

    at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1450)

    at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1438)

    at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1437)

    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:1437)

    at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:811)

    at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:811)

    at scala.Option.foreach(Option.scala:257)

    at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:811)

    at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1659)

    at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1618)

    at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1607)

    at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)

    at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:632)

    at org.apache.spark.SparkContext.runJob(SparkContext.scala:1871)

    at org.apache.spark.SparkContext.runJob(SparkContext.scala:1884)

    at org.apache.spark.SparkContext.runJob(SparkContext.scala:1897)

    at org.apache.spark.rdd.RDD$$anonfun$take$1.apply(RDD.scala:1305)

    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:358)

    at org.apache.spark.rdd.RDD.take(RDD.scala:1279)

    at org.apache.spark.streaming.dstream.DStream$$anonfun$print$2$$anonfun$foreachFunc$3$1.apply(DStream.scala:734)

    at org.apache.spark.streaming.dstream.DStream$$anonfun$print$2$$anonfun$foreachFunc$3$1.apply(DStream.scala:733)

    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:415)

    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:245)

    at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:245)

    at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:245)

    at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)

    at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:244)

    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)

    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)

    at java.lang.Thread.run(Thread.java:745)

Caused by: java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access

    at org.apache.kafka.clients.consumer.KafkaConsumer.acquire(KafkaConsumer.java:1430)

    at org.apache.kafka.clients.consumer.KafkaConsumer.seek(KafkaConsumer.java:1131)

    at org.apache.spark.streaming.kafka010.CachedKafkaConsumer.seek(CachedKafkaConsumer.scala:95)

    at org.apache.spark.streaming.kafka010.CachedKafkaConsumer.get(CachedKafkaConsumer.scala:69)

    at org.apache.spark.streaming.kafka010.KafkaRDD$KafkaRDDIterator.next(KafkaRDD.scala:227)

    at org.apache.spark.streaming.kafka010.KafkaRDD$KafkaRDDIterator.next(KafkaRDD.scala:193)

    at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)

    at scala.collection.Iterator$$anon$12.nextCur(Iterator.scala:434)

    at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:440)

    at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)

    at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:439)

    at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:408)

    at org.apache.spark.util.collection.ExternalSorter.insertAll(ExternalSorter.scala:192)

    at org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:63)

    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:79)

    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:47)

    at org.apache.spark.scheduler.Task.run(Task.scala:85)

    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)

    ... 3 more