title: How to work with JSON array as Kafka records in streaming queries (Java)?
body: <p>I have looked at many examples to read JSON data from a Kafa topic. I was able to do this successfully if I read a single record from the topic per connection, for example:</p>
  
  <pre><code>{"customer_id": "8d267162-1478-11ea-8d71-362b9e155667",
   "product": "Super widget",
   "price": 10,
   "bought_date": "2019-01-01"
  }
  </code></pre>
  
  <p>The code below works for the above use case:</p>
  
  <pre><code>package io.examle;
  
  import org.apache.spark.sql.Column;
  import org.apache.spark.sql.Dataset;
  import org.apache.spark.sql.Row;
  import org.apache.spark.sql.SparkSession;
  import org.apache.spark.sql.functions;
  import org.apache.spark.sql.streaming.StreamingQueryException;
  import org.apache.spark.sql.types.DataTypes;
  import org.apache.spark.sql.types.Metadata;
  import org.apache.spark.sql.types.StructField;
  import org.apache.spark.sql.types.StructType;
  
  public class Stackoverflow {
  
      public static void main(String[] args) throws StreamingQueryException {
  
          StructType schema = new StructType(new StructField[]{
                  new StructField("customer_id", DataTypes.StringType, false, Metadata.empty()),  
                  new StructField("product", DataTypes.StringType, false, Metadata.empty()),          
                  new StructField("price", DataTypes.IntegerType, false, Metadata.empty()),               
                  new StructField("bought_date", DataTypes.StringType, false, Metadata.empty()),
              });
  
          SparkSession spark = SparkSession
                  .builder()
                  .appName("SimpleExample")
                  .getOrCreate();
  
          // Create a DataSet representing the stream of input lines from Kafka
          Dataset&lt;Row&gt; dataset = spark
                          .readStream()
                          .format("kafka")                
                          .option("kafka.bootstrap.servers", "localhost:9092")
                          .option("subscribe", "utilization")
                          .load()
                          .selectExpr("CAST(value AS STRING) as json");
  
          dataset.printSchema();
  
          Column col = new Column("json");
  
          Dataset&lt;Row&gt; customers = dataset.select(functions.from_json(col,schema).as("data")).select("data.*");           
          customers.printSchema();
  
          customers.writeStream()      
          .format("console")
          .start()
          .awaitTermination();
  
      }
  
  }
  </code></pre>
  
  <p>But the above seems to me to be inefficient i.e making the connection to Kafa to get a single record per connection. So passing a JSON array of the form below, would be more efficient to my mind. As one can stock it with many "records" per json array.</p>
  
  <pre><code>[{
          "customer_id": "8d267162-1478-11ea-8d71-362b9e155667",
          "product": "Super widget",
          "price": 10,
          "bought_date": "2019-01-01"
      },
      {
          "customer_id": "498443a2-1479-11ea-8d71-362b9e155667",
          "product": "Food widget",
          "price": 4,
          "bought_date": "2019-01-01"
      } 
  ]
  </code></pre>
  
  <p>The problem is that I am unable to unpack the JSON Array and process it. The code below fails:</p>
  
  <pre><code>package io.example;
  
  import org.apache.spark.sql.Column;
  import org.apache.spark.sql.Dataset;
  import org.apache.spark.sql.Row;
  import org.apache.spark.sql.SparkSession;
  import org.apache.spark.sql.functions;
  import org.apache.spark.sql.streaming.StreamingQueryException;
  import org.apache.spark.sql.types.DataTypes;
  import org.apache.spark.sql.types.Metadata;
  import org.apache.spark.sql.types.StructField;
  import org.apache.spark.sql.types.StructType;
  
  public class Stackoverflow {
  
      public static void main(String[] args) throws StreamingQueryException {
  
          StructType schema = new StructType(new StructField[]{
                  new StructField("customer_id", DataTypes.StringType, false, Metadata.empty()),  
                  new StructField("product", DataTypes.StringType, false, Metadata.empty()),          
                  new StructField("price", DataTypes.IntegerType, false, Metadata.empty()),               
                  new StructField("bought_date", DataTypes.StringType, false, Metadata.empty()),
              });
  
          SparkSession spark = SparkSession
                  .builder()
                  .appName("SimpleExample")
                  .getOrCreate();
  
          // Create a DataSet representing the stream of input lines from Kafka
          Dataset&lt;Row&gt; dataset = spark
                          .readStream()
                          .format("kafka")                
                          .option("kafka.bootstrap.servers", "localhost:9092")
                          .option("subscribe", "utilization")
                          .load()
                          .selectExpr("CAST(value AS STRING) as json");
  
          dataset.printSchema();
  
          Column col = new Column("json");
  
          Dataset&lt;Row&gt; customers = dataset.select(functions.from_json(col,schema).as("data"));            
  
  
          Dataset&lt;Row&gt; data = customers.select(functions.explode_outer(functions.explode_outer(new Column("data"))));
          data.printSchema();
  
           data.writeStream()      
          .format("console")
          .start()
          .awaitTermination();
      }
  
  }
  
  
  
  
  Exception in thread "main" org.apache.spark.sql.AnalysisException: cannot resolve 'explode(`data`)' due to data type mismatch: input to function explode should be array or map type, not struct&lt;customer_id:string,product:string,price:int,bought_date:string&gt;;;
  </code></pre>
  
  <p>Questions:</p>
  
  <p>1) How to properly write code that will efficiently unpack the JSON array?  I doubt that the approach I took above for the code that fails is the best, but I tried to follow the many examples I saw regarding functions.explode() etc. </p>
  
  <p>2) If the code that fails is by some miracle a correct approach. How do I convert the struct to an array or a map?</p>