title: Spark Structured Streaming integrate with ElasticSearch
body: <p>I'm building an App which send some data to my cluster.
  I store this data into a specific HDFS folder, where a Spark Streaming application is running.</p>
  
  <p>In this streamApp, I'll do some fast &amp; cheap datascience.
  After, I have to index results to ElasticSearch to serve my AngularApp with data.</p>
  
  <p>Everything is working fine, </p>
  
  <p>BUT ... I can't index my results with ES.
  The fact is ... that I can't tranform my results DataFrame to RDD because it use some Dataframe Stream as input</p>
  
  <p>Here my pseudo-code :</p>
  
  <pre class="lang-scala prettyprint-override"><code>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 =&gt; 
          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
  </code></pre>
  
  <p>And It fail when I try to transform my DataFrame to RDD.</p>
  
  <p>I have to transform to RDD to be able to index data.</p>
  
  <p>On lines.rdd.map , I got this.</p>
  
  <pre><code>Exception in thread "main" org.apache.spark.sql.AnalysisException: Queries with streaming sources must be executed with writeStream.start();;
  </code></pre>
  
  <p>Is it possible to index DataStreaming spark inside ES ?</p>
  
  <p>Thanks for your help.</p>
  
  <p>Try more simple case :</p>
  
  <pre class="lang-scala prettyprint-override"><code>   val lines = spark.readStream
        .format("json")
        .schema(schema)
        .load(HDFSPATH).as[StreamingObj]
  
    lines.writeStream
        .format("org.elasticsearch.spark.sql")
        .outputMode("append")
        .start("index/stream")
  </code></pre>
  
  <blockquote>
    <p>17/12/21 15:37:55 INFO util.Version: Elasticsearch Hadoop v5.4.2
    [a478aabe9e] Exception in thread "main"
    java.lang.UnsupportedOperationException: Data source
    org.elasticsearch.spark.sql does not support streamed writing</p>
  </blockquote>
  
  <p>I do same thing as docs => <a href="https://www.elastic.co/guide/en/elasticsearch/hadoop/current/spark.html#spark-sql-streaming" rel="nofollow noreferrer">https://www.elastic.co/guide/en/elasticsearch/hadoop/current/spark.html#spark-sql-streaming</a></p>
  
  <p>or even this example : <a href="https://discuss.elastic.co/t/spark-structured-streaming-sink-in-append-mode/105664/4" rel="nofollow noreferrer">https://discuss.elastic.co/t/spark-structured-streaming-sink-in-append-mode/105664/4</a></p>
  
  <p>or this one :</p>
  
  <p><a href="https://discuss.elastic.co/t/structured-streaming-failed-to-find-data-source-es/112144" rel="nofollow noreferrer">https://discuss.elastic.co/t/structured-streaming-failed-to-find-data-source-es/112144</a></p>
  
  <p>Here my Maven dependency :</p>
  
  <pre><code>&lt;dependency&gt;
      &lt;groupId&gt;org.elasticsearch&lt;/groupId&gt;
      &lt;artifactId&gt;elasticsearch-spark-20_2.11&lt;/artifactId&gt;
      &lt;version&gt;5.4.2&lt;/version&gt;
  &lt;/dependency&gt;
  </code></pre>
  
  <p>Is it the good one ? I can't use format('es') , it densest find it.</p>
  
  <p>Seems that Spark Structured streaming in ES is only > 6.0</p>
  
  <p>See <a href="https://www.elastic.co/blog/structured-streaming-elasticsearch-for-hadoop-6-0" rel="nofollow noreferrer">https://www.elastic.co/blog/structured-streaming-elasticsearch-for-hadoop-6-0</a></p>