我目前正在编写一个由生产者和消费者组成的scala应用程序。制作人从外部获取一些数据，并将它们写入Kafka内部。消费者阅读Kafka的内容，然后写信给elasticsearch。
消费者基于spark流媒体，每5秒从kafka获取新消息并将其写入elasticsearch。问题是我无法写入es，因为我会遇到很多错误，如以下错误：
错误][2015-04-24 11:21:14734][org.apache.spark.taskcontextimpl]：taskcompletionlistener org.elasticsearch.hadoop.eshadoopexception中出错：无法写入所有条目[3/26560]（可能es过载？）。救出。。。在org.elasticsearch.hadoop.rest.restrepository.flush（restrepository。java:225)~[elasticsearch-spark\ 2.10-2.1.0.beta3。jar:2.1.0.beta3]在org.elasticsearch.hadoop.rest.restrepository.close（restrepository。java:236)~[elasticsearch-spark\ 2.10-2.1.0.beta3。jar:2.1.0.beta3]在org.elasticsearch.hadoop.rest.restservice$partitionwriter.close（restservice。java:125)~[elasticsearch-spark\ 2.10-2.1.0.beta3。jar:2.1.0.beta3]在org.elasticsearch.spark.rdd.esrddriter$$anonfun$write$1.apply$mcv$sp（esrddriter。scala:33)~[elasticsearch-spark\ 2.10-2.1.0.beta3。jar:2.1.0.beta3]在org.apache.spark.taskcontextimpl$$anon$2.ontaskcompletion（taskcontextimpl。scala:57)[Spark芯2.10-1.2.1。jar:1.2.1]在org.apache.spark.taskcontextimpl$$anonfun$marktaskcompleted$1.apply（taskcontextimpl。scala:68)[Spark芯2.10-1.2.1。jar:1.2.1]在org.apache.spark.taskcontextimpl$$anonfun$marktaskcompleted$1.apply（taskcontextimpl。scala:66) [Spark芯2.10-1.2.1。jar:1.2.1]在scala.collection.mutable.resizablearray$class.foreach（resizablearray。scala:59)[na:na]在scala.collection.mutable.arraybuffer.foreach（arraybuffer。scala:47)[na:na]位于org.apache.spark.taskcontextimpl.marktaskcompleted（taskcontextimpl。scala:66)[Spark芯2.10-1.2.1。jar:1.2.1]在org.apache.spark.scheduler.task.run（任务。scala:58)[Spark芯2.10-1.2.1。jar:1.2.1]在org.apache.spark.executor.executor$taskrunner.run（executor。scala:200)[Spark芯2.10-1.2.1。jar:1.2.1]位于java.util.concurrent.threadpoolexecutor.runworker（threadpoolexecutor。java:1145)[na:1.7.0_]在java.util.concurrent.threadpoolexecutor$worker.run（threadpoolexecutor。java:615)[na:1.7.0_]at java.lang.thread.run（线程。java:745)[na:1.7.0ｷ65]
假设制作人每15秒写6条消息，所以我真的不明白这种“过载”是怎么发生的（我甚至清理了主题并刷新了所有旧消息，我认为这与偏移量问题有关）。spark streaming每5秒执行一次的任务可以用以下代码总结：

val result = KafkaUtils.createStream[String, Array[Byte], StringDecoder, DefaultDecoder](ssc, kafkaParams, Map("wasp.raw" -> 1), StorageLevel.MEMORY_ONLY_SER_2)
  val convertedResult = result.map(k => (k._1 ,AvroToJsonUtil.avroToJson(k._2)))

  //TO-DO : Remove resource (yahoo/yahoo) hardcoded parameter
  log.info(s"***EXECUTING SPARK STREAMING TASK  + ${java.lang.System.currentTimeMillis()}***")

  convertedResult.foreachRDD(rdd => {
      rdd.map(data => data._2).saveToEs("yahoo/yahoo", Map("es.input.json" -> "true"))

  })

如果我尝试打印消息而不是发送给es，一切都很好，实际上我只看到6条消息。为什么我不能给她写信？
为了完整起见，我使用这个库编写了es:elasticsearch-spark_.10的最新beta版本。