title: Spark Streaming Kafka to ES
body: <p>I have a spark streaming job, which will read from kafka and write into elastic via Http request.</p>
  <p>I want to validate each request from Kafka and change the payload as per business need and write into Elastic Search.</p>
  <p>I have used ES Http Request to push the data into Elastic Search. Can some guide me how to write the data into ES via a data frame?</p>
  <p>Code Snippet:</p>
  <pre><code>val dfInput = spark
    .readStream
    .format(&quot;kafka&quot;)
    .option(&quot;kafka.bootstrap.servers&quot;, &quot;localhost:9092&quot;)
    .option(&quot;subscribe&quot;, &quot;test&quot;)
    .option(&quot;startingOffsets&quot;, &quot;latest&quot;)
    .option(&quot;group.id&quot;, sourceTopicGroupId)
    .option(&quot;failOnDataLoss&quot;, &quot;false&quot;)
    .option(&quot;maxOffsetsPerTrigger&quot;, maxOffsetsPerTrigger)
    .load()
  
  import spark.implicits._
  
  val resultDf = dfInput
    .withColumn(&quot;value&quot;, $&quot;value&quot;.cast(&quot;string&quot;))
    .select(&quot;value&quot;)
  
  resultDf.writeStream.foreach(new ForeachWriter[Row] {
    override def open(partitionId: Long, version: Long): Boolean = true
  
    override def process(value: Row): Unit = {
      processEventsData(value.get(0).asInstanceOf[String], deviceIndex, msgIndex, retryOnConflict,auth,refreshInterval,deviceUrl,messageUrl,spark)
    }
  
    override def close(errorOrNull: Throwable): Unit = {
    }
  }).trigger(Trigger.ProcessingTime(triggerPeriod)).start().awaitTermination() //&quot;1 second&quot;
  }
  </code></pre>
  <p>This way we are not able to achieve performance.</p>
  <p>Is there any approach?</p>
  <ul>
  <li>Spark Version 2.3.2</li>
  <li>Kafka Partitions 20</li>
  <li>ES version 7.7.0</li>
  </ul>