Spark Structured Streaming 与 ElasticSearch 集成答案

【问题标题】：Spark Structured Streaming integrate with ElasticSearchSpark Structured Streaming 与 ElasticSearch 集成
【发布时间】：2018-06-04 04:51:10
【问题描述】：

我正在构建一个向我的集群发送一些数据的应用程序。我将这些数据存储到运行 Spark Streaming 应用程序的特定 HDFS 文件夹中。

在这个 streamApp 中，我将做一些快速且廉价的数据科学。之后，我必须将结果索引到 ElasticSearch 以便为我的 AngularApp 提供数据。

一切正常，

但是...我无法用 ES 索引我的结果。事实是......我无法将我的结果 DataFrame 转换为 RDD，因为它使用一些 Dataframe Stream 作为输入

这是我的伪代码：

val schema = StructType(
    StructField("id", StringType, nullable = false) ::
    StructField("code", StringType, nullable = false) :: Nil)


val lines = spark.readStream
  .format("json")
  .schema(schema)
  .load(HDFSPath)

// Do some basics stuff here

import spark.implicits._

val linesRDD = lines.rdd.map(row => 
        StreamingObj(row(0).toString,row(1).toString))  // RDD[StreamingObj]
linesRDD.saveToEs("stream/stream")           // ES
val linesDF= linesRDD.toDF()

val queryNode = linesDF
    .writeStream
    .format("console")
    .outputMode(OutputMode.Append)
    .trigger(Trigger.ProcessingTime(4.seconds))
    .start

当我尝试将我的 DataFrame 转换为 RDD 时它失败了。

我必须转换为 RDD 才能索引数据。

在lines.rdd.map 上，我得到了这个。

Exception in thread "main" org.apache.spark.sql.AnalysisException: Queries with streaming sources must be executed with writeStream.start();;

是否可以在 ES 中索引 DataStreaming spark ？

感谢您的帮助。

尝试更简单的案例：

   val lines = spark.readStream
      .format("json")
      .schema(schema)
      .load(HDFSPATH).as[StreamingObj]

  lines.writeStream
      .format("org.elasticsearch.spark.sql")
      .outputMode("append")
      .start("index/stream")

17/12/21 15:37:55 信息 util.Version：Elasticsearch Hadoop v5.4.2 [a478aabe9e] 线程“main”中的异常 java.lang.UnsupportedOperationException：数据源 org.elasticsearch.spark.sql 不支持流式写入

我和 docs 做同样的事情 => https://www.elastic.co/guide/en/elasticsearch/hadoop/current/spark.html#spark-sql-streaming

甚至这个例子：https://discuss.elastic.co/t/spark-structured-streaming-sink-in-append-mode/105664/4

或者这个：

https://discuss.elastic.co/t/structured-streaming-failed-to-find-data-source-es/112144

这是我的 Maven 依赖项：

<dependency>
    <groupId>org.elasticsearch</groupId>
    <artifactId>elasticsearch-spark-20_2.11</artifactId>
    <version>5.4.2</version>
</dependency>

这是一个好人吗？我不能使用 format('es') ，它最密集的找到它。

似乎 ES 中的 Spark Structured 流式传输仅 > 6.0

见https://www.elastic.co/blog/structured-streaming-elasticsearch-for-hadoop-6-0

【问题讨论】：

标签： apache-spark elasticsearch spark-structured-streaming

【解决方案1】：

问题来了

val linesRDD = lines.rdd.map(row => 
        StreamingObj(row(0).toString,row(1).toString))  // RDD[StreamingObj]

结构化流式查询中不允许转换为 RDD。你可以尝试直接写：

lines
  .writeStream
  .format("org.elasticsearch.spark.sql")
  ...

或使用ForeachWriter:

lines.writeStream.foreach(...)

【讨论】：

【解决方案2】：

要在 Elasticsearch 5.x 中实现这一点，您需要像在 this answer 中那样实现 CustomSink 和 CustomSinkProvider。

然后在lines.writeStream.foreach(...) 中您可以在format 中指定提供程序。

【讨论】：