How to transform array of JSONs to rows before writeStream to Elasticsearch?












0















Follow-up to this question



I have JSON streaming data in the format same as below



|  A    | B                                        |
|-------|------------------------------------------|
| ABC | [{C:1, D:1}, {C:2, D:4}] |
| XYZ | [{C:3, D :6}, {C:9, D:11}, {C:5, D:12}] |


I need to transform it to the format below



|   A   |  C  |  D   |
|-------|-----|------|
| ABC | 1 | 1 |
| ABC | 2 | 4 |
| XYZ | 3 | 6 |
| XYZ | 9 | 11 |
| XYZ | 5 | 12 |


To achieve this performed the transformations as suggested to the previous question.



val df1 = df0.select($"A", explode($"B")).toDF("A", "Bn")

val df2 = df1.withColumn("SeqNum", monotonically_increasing_id()).toDF("A", "Bn", "SeqNum")

val df3 = df2.select($"A", explode($"Bn"), $"SeqNum").toDF("A", "B", "C", "SeqNum")

val df4 = df3.withColumn("dummy", concat( $"SeqNum", lit("||"), $"A"))

val df5 = df4.select($"dummy", $"B", $"C").groupBy("dummy").pivot("B").agg(first($"C"))

val df6 = df5.withColumn("A", substring_index(col("dummy"), "||", -1)).drop("dummy")


Now I need to save the data to a ElasticSearch.



 df6.writeStream
.outputMode("complete")
.format("es")
.option("es.resource", "index/type")
.option("es.nodes", "localhost")
.option("es.port", 9200)
.start()
.awaitTermination()


I get an error that ElasticSearch doesn't support Append output mode. On Append mode it fails write to writeStream with aggregation cannot be done on Append mode. I was able to write to console on complete mode. How can I write the data to ElasticSearch now



Any help will be appreciated.










share|improve this question





























    0















    Follow-up to this question



    I have JSON streaming data in the format same as below



    |  A    | B                                        |
    |-------|------------------------------------------|
    | ABC | [{C:1, D:1}, {C:2, D:4}] |
    | XYZ | [{C:3, D :6}, {C:9, D:11}, {C:5, D:12}] |


    I need to transform it to the format below



    |   A   |  C  |  D   |
    |-------|-----|------|
    | ABC | 1 | 1 |
    | ABC | 2 | 4 |
    | XYZ | 3 | 6 |
    | XYZ | 9 | 11 |
    | XYZ | 5 | 12 |


    To achieve this performed the transformations as suggested to the previous question.



    val df1 = df0.select($"A", explode($"B")).toDF("A", "Bn")

    val df2 = df1.withColumn("SeqNum", monotonically_increasing_id()).toDF("A", "Bn", "SeqNum")

    val df3 = df2.select($"A", explode($"Bn"), $"SeqNum").toDF("A", "B", "C", "SeqNum")

    val df4 = df3.withColumn("dummy", concat( $"SeqNum", lit("||"), $"A"))

    val df5 = df4.select($"dummy", $"B", $"C").groupBy("dummy").pivot("B").agg(first($"C"))

    val df6 = df5.withColumn("A", substring_index(col("dummy"), "||", -1)).drop("dummy")


    Now I need to save the data to a ElasticSearch.



     df6.writeStream
    .outputMode("complete")
    .format("es")
    .option("es.resource", "index/type")
    .option("es.nodes", "localhost")
    .option("es.port", 9200)
    .start()
    .awaitTermination()


    I get an error that ElasticSearch doesn't support Append output mode. On Append mode it fails write to writeStream with aggregation cannot be done on Append mode. I was able to write to console on complete mode. How can I write the data to ElasticSearch now



    Any help will be appreciated.










    share|improve this question



























      0












      0








      0








      Follow-up to this question



      I have JSON streaming data in the format same as below



      |  A    | B                                        |
      |-------|------------------------------------------|
      | ABC | [{C:1, D:1}, {C:2, D:4}] |
      | XYZ | [{C:3, D :6}, {C:9, D:11}, {C:5, D:12}] |


      I need to transform it to the format below



      |   A   |  C  |  D   |
      |-------|-----|------|
      | ABC | 1 | 1 |
      | ABC | 2 | 4 |
      | XYZ | 3 | 6 |
      | XYZ | 9 | 11 |
      | XYZ | 5 | 12 |


      To achieve this performed the transformations as suggested to the previous question.



      val df1 = df0.select($"A", explode($"B")).toDF("A", "Bn")

      val df2 = df1.withColumn("SeqNum", monotonically_increasing_id()).toDF("A", "Bn", "SeqNum")

      val df3 = df2.select($"A", explode($"Bn"), $"SeqNum").toDF("A", "B", "C", "SeqNum")

      val df4 = df3.withColumn("dummy", concat( $"SeqNum", lit("||"), $"A"))

      val df5 = df4.select($"dummy", $"B", $"C").groupBy("dummy").pivot("B").agg(first($"C"))

      val df6 = df5.withColumn("A", substring_index(col("dummy"), "||", -1)).drop("dummy")


      Now I need to save the data to a ElasticSearch.



       df6.writeStream
      .outputMode("complete")
      .format("es")
      .option("es.resource", "index/type")
      .option("es.nodes", "localhost")
      .option("es.port", 9200)
      .start()
      .awaitTermination()


      I get an error that ElasticSearch doesn't support Append output mode. On Append mode it fails write to writeStream with aggregation cannot be done on Append mode. I was able to write to console on complete mode. How can I write the data to ElasticSearch now



      Any help will be appreciated.










      share|improve this question
















      Follow-up to this question



      I have JSON streaming data in the format same as below



      |  A    | B                                        |
      |-------|------------------------------------------|
      | ABC | [{C:1, D:1}, {C:2, D:4}] |
      | XYZ | [{C:3, D :6}, {C:9, D:11}, {C:5, D:12}] |


      I need to transform it to the format below



      |   A   |  C  |  D   |
      |-------|-----|------|
      | ABC | 1 | 1 |
      | ABC | 2 | 4 |
      | XYZ | 3 | 6 |
      | XYZ | 9 | 11 |
      | XYZ | 5 | 12 |


      To achieve this performed the transformations as suggested to the previous question.



      val df1 = df0.select($"A", explode($"B")).toDF("A", "Bn")

      val df2 = df1.withColumn("SeqNum", monotonically_increasing_id()).toDF("A", "Bn", "SeqNum")

      val df3 = df2.select($"A", explode($"Bn"), $"SeqNum").toDF("A", "B", "C", "SeqNum")

      val df4 = df3.withColumn("dummy", concat( $"SeqNum", lit("||"), $"A"))

      val df5 = df4.select($"dummy", $"B", $"C").groupBy("dummy").pivot("B").agg(first($"C"))

      val df6 = df5.withColumn("A", substring_index(col("dummy"), "||", -1)).drop("dummy")


      Now I need to save the data to a ElasticSearch.



       df6.writeStream
      .outputMode("complete")
      .format("es")
      .option("es.resource", "index/type")
      .option("es.nodes", "localhost")
      .option("es.port", 9200)
      .start()
      .awaitTermination()


      I get an error that ElasticSearch doesn't support Append output mode. On Append mode it fails write to writeStream with aggregation cannot be done on Append mode. I was able to write to console on complete mode. How can I write the data to ElasticSearch now



      Any help will be appreciated.







      apache-spark elasticsearch spark-structured-streaming elasticsearch-spark






      share|improve this question















      share|improve this question













      share|improve this question




      share|improve this question








      edited Nov 25 '18 at 19:53









      Jacek Laskowski

      46.2k18137276




      46.2k18137276










      asked Nov 23 '18 at 11:42









      Hasif SubairHasif Subair

      1222318




      1222318
























          1 Answer
          1






          active

          oldest

          votes


















          1














          There is no need for pivot or aggregation here. If B column is indeed Array[Map[String, String]] (array<map<string, string>> in SQL types), all you need is a simple select or withColumn:



          df
          .withColumn("B", explode($"B"))
          .select($"A", $"B"("C") as "C", $"B"("D") as "D")





          share|improve this answer
























            Your Answer






            StackExchange.ifUsing("editor", function () {
            StackExchange.using("externalEditor", function () {
            StackExchange.using("snippets", function () {
            StackExchange.snippets.init();
            });
            });
            }, "code-snippets");

            StackExchange.ready(function() {
            var channelOptions = {
            tags: "".split(" "),
            id: "1"
            };
            initTagRenderer("".split(" "), "".split(" "), channelOptions);

            StackExchange.using("externalEditor", function() {
            // Have to fire editor after snippets, if snippets enabled
            if (StackExchange.settings.snippets.snippetsEnabled) {
            StackExchange.using("snippets", function() {
            createEditor();
            });
            }
            else {
            createEditor();
            }
            });

            function createEditor() {
            StackExchange.prepareEditor({
            heartbeatType: 'answer',
            autoActivateHeartbeat: false,
            convertImagesToLinks: true,
            noModals: true,
            showLowRepImageUploadWarning: true,
            reputationToPostImages: 10,
            bindNavPrevention: true,
            postfix: "",
            imageUploader: {
            brandingHtml: "Powered by u003ca class="icon-imgur-white" href="https://imgur.com/"u003eu003c/au003e",
            contentPolicyHtml: "User contributions licensed under u003ca href="https://creativecommons.org/licenses/by-sa/3.0/"u003ecc by-sa 3.0 with attribution requiredu003c/au003e u003ca href="https://stackoverflow.com/legal/content-policy"u003e(content policy)u003c/au003e",
            allowUrls: true
            },
            onDemand: true,
            discardSelector: ".discard-answer"
            ,immediatelyShowMarkdownHelp:true
            });


            }
            });














            draft saved

            draft discarded


















            StackExchange.ready(
            function () {
            StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f53446051%2fhow-to-transform-array-of-jsons-to-rows-before-writestream-to-elasticsearch%23new-answer', 'question_page');
            }
            );

            Post as a guest















            Required, but never shown

























            1 Answer
            1






            active

            oldest

            votes








            1 Answer
            1






            active

            oldest

            votes









            active

            oldest

            votes






            active

            oldest

            votes









            1














            There is no need for pivot or aggregation here. If B column is indeed Array[Map[String, String]] (array<map<string, string>> in SQL types), all you need is a simple select or withColumn:



            df
            .withColumn("B", explode($"B"))
            .select($"A", $"B"("C") as "C", $"B"("D") as "D")





            share|improve this answer




























              1














              There is no need for pivot or aggregation here. If B column is indeed Array[Map[String, String]] (array<map<string, string>> in SQL types), all you need is a simple select or withColumn:



              df
              .withColumn("B", explode($"B"))
              .select($"A", $"B"("C") as "C", $"B"("D") as "D")





              share|improve this answer


























                1












                1








                1







                There is no need for pivot or aggregation here. If B column is indeed Array[Map[String, String]] (array<map<string, string>> in SQL types), all you need is a simple select or withColumn:



                df
                .withColumn("B", explode($"B"))
                .select($"A", $"B"("C") as "C", $"B"("D") as "D")





                share|improve this answer













                There is no need for pivot or aggregation here. If B column is indeed Array[Map[String, String]] (array<map<string, string>> in SQL types), all you need is a simple select or withColumn:



                df
                .withColumn("B", explode($"B"))
                .select($"A", $"B"("C") as "C", $"B"("D") as "D")






                share|improve this answer












                share|improve this answer



                share|improve this answer










                answered Nov 23 '18 at 11:54









                user10465355user10465355

                2,1192521




                2,1192521
































                    draft saved

                    draft discarded




















































                    Thanks for contributing an answer to Stack Overflow!


                    • Please be sure to answer the question. Provide details and share your research!

                    But avoid



                    • Asking for help, clarification, or responding to other answers.

                    • Making statements based on opinion; back them up with references or personal experience.


                    To learn more, see our tips on writing great answers.




                    draft saved


                    draft discarded














                    StackExchange.ready(
                    function () {
                    StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f53446051%2fhow-to-transform-array-of-jsons-to-rows-before-writestream-to-elasticsearch%23new-answer', 'question_page');
                    }
                    );

                    Post as a guest















                    Required, but never shown





















































                    Required, but never shown














                    Required, but never shown












                    Required, but never shown







                    Required, but never shown

































                    Required, but never shown














                    Required, but never shown












                    Required, but never shown







                    Required, but never shown







                    這個網誌中的熱門文章

                    Xamarin.form Move up view when keyboard appear

                    Post-Redirect-Get with Spring WebFlux and Thymeleaf

                    Anylogic : not able to use stopDelay()