# Spark Structured Streaming JSON Serialization

**URL:** <https://discuss.elastic.co/t/spark-structured-streaming-json-serialization/214661>\
**Category:** Elasticsearch\
**Tags:** es-hadoop\
**Created:** [January 10, 2020, 8:37pm UTC](https://discuss.elastic.co/t/spark-structured-streaming-json-serialization/214661 "2020-01-10T20:37:51Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![execdd17](https://avatars.discourse-cdn.com/v4/letter/e/85f322/32.png) [@execdd17](https://discuss.elastic.co/u/execdd17)\
**Post date:** [January 10, 2020, 8:37pm UTC](https://discuss.elastic.co/t/spark-structured-streaming-json-serialization/214661/1 "2020-01-10T20:37:51Z")

</div>

Hey Everyone,

I hope this question hasn't been asked before, I looked around quite a bit before posting. If I missed it, sorry!

I'm trying to use the `org.elasticsearch.spark.sql` structured streaming sink for ES, and am running into some challenges when my column contains a string literal of valid JSON. I have one column called `combined`.

When the documents are written to ES they look like this:

```
"_source" : {
          "combined" : """{"sparkTimestamp":"2020-01-08T15:42:24.890Z","@timestamp":"2020-01-08T15:39:40.583Z","@metadata":{"beat":"winlogbeat","type":"_doc","version":"7.4.2","topic":"ingest-winlogbeat"},"host":{"hostname":"A HOSTNAME"................ truncated for brevity

```

As you can see, it's assuming that I have a normal string, and is nice enough to escape it for me too. 😄

What I am really after though, is for the JSON string within the `combined` column to be the `_source`. It looks like this is possible with the older RDD approach, but I didn't see any way to make it work within the context of this sink. Is this possible? If there is another way, what are the message delivery guarantees?

If you're wondering _why_ I'm doing this, then I'm happy to briefly explain. I'm using winlogbeat as a source, and while it is ECS compliant, a vast number of fields are within a custom extension. In addition, depending on the event, the fields will change. If I try to account for this in a spark managed(SQL) table, it's going to be very difficult because of the sparsity and sheer number of fields.

To alleviate this I parsed out core ECS fields and retained the original JSON from winlogbeat. Now that the raw table has been established, I want to index the original JSON into ES. This is effectively the `combined` column.

Any insight would be appreciated. Thanks!

---

<div class="post-metadata">

**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:** [January 28, 2020, 9:48pm UTC](https://discuss.elastic.co/t/spark-structured-streaming-json-serialization/214661/2 "2020-01-28T21:48:59Z")

</div>

I don't think this is a supported write configuration in ES Hadoop at the moment. We convert all Dataset objects into Dataframe types before consuming them. I think it's reasonable that we should be supporting this method if ingest - I believe that the Hive integration is able to support it, but in a somewhat more clumsy fashion.

One thing you might want to try: When you are done transforming your data and are ready to index the Dataset, try converting the Dataset[String] into an RDD[String] and saving it to Elasticsearch that way instead? I know that we do support this sort of write operation with RDD's so hopefully that will be good enough to get you past this hurdle.

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [February 25, 2020, 9:49pm UTC](https://discuss.elastic.co/t/spark-structured-streaming-json-serialization/214661/3 "2020-02-25T21:49:09Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.