如何确保在Spark Streaming中使用Elasticsearch-Hadoop连接器写入Elasticsearch集成的所有文档

Question

我正在使用Elasticsearch-Hadoop连接器将DStream写到Elasticsearch。这是您可以找到的连接器https://www.elastic.co/guide/en/elasticsearch/hadoop/5.6/spark.html的链接

我需要处理窗口，使用"JavaEsSpark.saveToEs“方法将所有文档写入ES，并希望确保所有写入的文档和提交到Kafka的偏移量。由于JavaEsSpark.saveToEs以批处理模式插入文档，因此我无法跟踪我的文档。

我的基本代码如下。有什么意见吗？

    dstream.foreachRDD((items, time) -> {
        JavaEsSpark.saveToEs(items,"myindex/mytype");
        //wait until all the documents written
        //do somehing else then return (actually the job is committing kafka offsets)
});

Answer 1

您可以将您的函数封装在Try中(这是一个Scala示例)：

 Try {
  rdd.saveToEs(AppSettings.Elastic.Resource, configuration)
} match {
  case Failure(f) =>
    logger.error(s"SaveToEs failed: $f") //or whatever you want
  case _ =>
}