# Spark Structured Streaming - sink in append mode **URL:** \ **Category:** Elasticsearch\ **Tags:** es-hadoop\ **Created:** [October 29, 2017, 2:06pm UTC](https://discuss.elastic.co/t/spark-structured-streaming-sink-in-append-mode/105664 "2017-10-29T14:06:22Z")\ **Posts on this page:** 6\ **Page:** 1
**Author:** ![hcloli](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/hcloli/32/23549_2.png) [@hcloli](https://discuss.elastic.co/u/hcloli)\ **Post date:** [October 29, 2017, 2:06pm UTC](https://discuss.elastic.co/t/spark-structured-streaming-sink-in-append-mode/105664/1 "2017-10-29T14:06:23Z")
Hi James, Great job regarding support for Spark 2.2.0 structured streaming!! I tried it and it works well. However, I wonder why you limited the sink to work only in APPEND mode. If user provides **[es.mapping.id](http://es.mapping.id)** column, you can support UPDATE mode easily (actually, with no code change at all). When information for a certain window of time arrives, the sink will write the data to elasticsearch using the ID. When an update arrives for the same ID, elasticsearch will overwrite that document, since the same ID was used. Adding the ability to work in UPDATE mode will make elasticsearch a great selection for a sink in our project. BTW, when is it plan to make elasticsearch-hadoop 6.0.0 a GA version? Many Thanks, Haim ---
**Author:** ![james.baiera](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/james.baiera/32/10209_2.png) [@james.baiera](https://discuss.elastic.co/u/james.baiera)\ **Post date:** [October 31, 2017, 6:55pm UTC](https://discuss.elastic.co/t/spark-structured-streaming-sink-in-append-mode/105664/2 "2017-10-31T18:55:16Z")
Thanks for trying out the new structured streaming feature and for your feedback on it! We agree on the UPDATE mode sentiments and have plans to eventually support it in a later release. We wanted to make sure that we were supporting what we felt was the most common use case in the first iteration. Keep an eye open though! ---
**Author:** ![mans4singh](https://avatars.discourse-cdn.com/v4/letter/m/a183cd/32.png) [@mans4singh](https://discuss.elastic.co/u/mans4singh)\ **Post date:** [November 16, 2017, 3:54am UTC](https://discuss.elastic.co/t/spark-structured-streaming-sink-in-append-mode/105664/3 "2017-11-16T03:54:35Z")
Hi Haim: I am trying to use ES Sink for Structured streaming but did not find any working example. I tried using the example [https://www.elastic.co/guide/en/elasticsearch/hadoop/current/spark.html#spark-sql-streaming](https://www.elastic.co/guide/en/elasticsearch/hadoop/current/spark.html#spark-sql-streaming) in a test application with socket as source and it seems to hang. If you've been able to use it, can you please share an sample/documentation on it's usage ? Thanks ---
**Author:** ![hcloli](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/hcloli/32/23549_2.png) [@hcloli](https://discuss.elastic.co/u/hcloli)\ **Post date:** [November 16, 2017, 12:03pm UTC](https://discuss.elastic.co/t/spark-structured-streaming-sink-in-append-mode/105664/4 "2017-11-16T12:03:53Z")
Hi Mans, It works really well for me. Several considerations: 1. End the program with > query.awaitTermination(); 1. Configure **checkpoint** location. 2. Output mode must be **append**. 3. In **SparkSession** you must configure elasticsearch details. 4. In groupBy datasets you must configure **watermark**. 5. The **start** method on the query should contain the index name and doc type. 6. Maybe your logging definition is not set correctly so you cannot see Spark's errors? All the configurations above create exceptions if not set correctly. 7. **Spark 2.2.0** is required. Here is my full java code for handling CSV files. I have other program that create those files every 10 seconds and add them to the input dir. I ran it both on my local machine and in spark cluster (had to remove the **master** declaration from SparkSession to let spark-submit.sh define it). ```java import org.apache.spark.sql.*; import org.apache.spark.sql.streaming.StreamingQuery; import org.apache.spark.sql.types.*; import org.elasticsearch.hadoop.cfg.ConfigurationOptions; import static org.apache.spark.sql.functions.*; public class SparkJavaElasticStreamTest { public static void main(String[] args) { try { SparkSession spark = SparkSession.builder() .config(ConfigurationOptions.ES_NET_HTTP_AUTH_USER, "elastic") .config(ConfigurationOptions.ES_NET_HTTP_AUTH_PASS, "somepass") .config(ConfigurationOptions.ES_NODES, "127.0.0.1") .config(ConfigurationOptions.ES_PORT, "9200") .appName("StreamingElastic") .master("local[*]") .getOrCreate(); StructType schema = new StructType(new StructField[]{ new StructField("batch", DataTypes.StringType, false, Metadata.empty()) ,new StructField("key", DataTypes.StringType, false, Metadata.empty()) ,new StructField("time", DataTypes.TimestampType, false, Metadata.empty()) ,new StructField("amount", DataTypes.LongType, false, Metadata.empty()) }); Dataset file = spark .readStream() .schema(schema) .csv("/tmp/stream1"); Dataset grouped = file .withWatermark("time", "1 seconds") .groupBy(window(col("time"), "5 minutes", "1 minute")) .agg(last(col("batch")).alias("last_batch") , count(col("*")).alias("count5") , sum("amount").alias("sumAmount5") , avg("amount").alias("avgAmount5") , stddev("amount").alias("stddevAmount5") , min("amount").alias("minAmount5") , max("amount").alias("maxAmount5")); StreamingQuery query = grouped .writeStream() .outputMode("append") .format("es") .option("checkpointLocation", "/tmp/stream1_checkpoint") .start("index1/type1"); query.awaitTermination(); } catch (Exception e) { e.printStackTrace(); } } } ``` ---
**Author:** ![mans4singh](https://avatars.discourse-cdn.com/v4/letter/m/a183cd/32.png) [@mans4singh](https://discuss.elastic.co/u/mans4singh)\ **Post date:** [November 17, 2017, 3:13am UTC](https://discuss.elastic.co/t/spark-structured-streaming-sink-in-append-mode/105664/5 "2017-11-17T03:13:17Z")
Thanks Haim for the sample code. I will try that. Mans ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [December 15, 2017, 3:13am UTC](https://discuss.elastic.co/t/spark-structured-streaming-sink-in-append-mode/105664/6 "2017-12-15T03:13:25Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.