title: apache spark streaming with kafka and hive integration
body: <p>Is there anyway to integrate apache spark structured streaming with apache hive and apache kafka in one application</p>
  
  <p>after adding list using collectAsList and storing it into list. I got the below error.</p>
  
  <p>Can anyone help me to resolve this issues.</p>
  
  <p>Thanks in Advance.</p>
  
  <pre><code>import org.apache.spark.api.java.function.MapFunction;
  
  import java.io.IOException;
  import java.util.Arrays;
  import java.util.List;
  
  import org.apache.log4j.Level;
  import org.apache.log4j.Logger;
  import org.apache.spark.sql.Dataset;
  import org.apache.spark.sql.Encoders;
  import org.apache.spark.sql.Row;
  import org.apache.spark.sql.SparkSession;
  import org.apache.spark.sql.streaming.StreamingQuery;
  import org.apache.spark.sql.streaming.StreamingQueryException;
  
  
  public class DatasetKafka {
      public static void main(String[] args) throws IOException {
          SparkSession spark = SparkSession
                  .builder()
                  .appName("Java Spark Hive Example").master("yarn")
                  .config("spark.sql.warehouse.dir", "hdfs://localhost:54310/user/hive/warehouse")
                  .enableHiveSupport()
                  .getOrCreate();
          Logger.getRootLogger().setLevel(Level.ERROR);
          Dataset&lt;String&gt; lines = spark
                    .readStream()
                    .format("kafka")
                    .option("kafka.bootstrap.servers", "localhost:9092")
                    .option("subscribe", "test")
                    .load().selectExpr("CAST(value AS STRING)")
                    .as(Encoders.STRING());
          List&lt;String&gt; line=lines.collectAsList();
          for(String li:line) {
              String values[]=li.split(",");
              String query="insert into table match values("+Integer.parseInt(
              values[0])+
              ","+values[1]+
              ","+Integer.parseInt(values[2])+
              ","+Integer.parseInt(values[3])+
              ","+Integer.parseInt(values[4])+
              ","+values[5]+
              ","+Integer.parseInt(values[6])+
              ","+values[7]+
              ","+Integer.parseInt(values[8])+
              ","+Integer.parseInt(values[9])+
              ","+Integer.parseInt(values[10])+
              ","+values[11]+
              ","+Integer.parseInt(values[12])+
              ","+Integer.parseInt(values[13])+
              ","+Integer.parseInt(values[14])+
              ","+Integer.parseInt(values[15])+
              ","+Integer.parseInt(values[16])+
              ","+values[17]+
              ","+values[18]+")";
              spark.sql(query);
          }
  
  //      List&lt;String&gt; values=ll.collectAsList();
          Dataset&lt;String&gt; words=lines.map((MapFunction&lt;String, String&gt;)k-&gt;{
              return k;
          }, Encoders.STRING());
          Dataset&lt;Row&gt; wordCounts = words.flatMap(
                  (FlatMapFunction&lt;String, String&gt;) x -&gt; Arrays.asList(x.split(",")).iterator(),
                  Encoders.STRING()).groupBy("value").count();
          StreamingQuery query = wordCounts.writeStream()
                    .outputMode("complete")
                    .format("console")
                    .start();
                  try {
                      query.awaitTermination();
                  } catch (StreamingQueryException e) {
                      // TODO Auto-generated catch block
                      e.printStackTrace();
                  }
      }
  </code></pre>
  
  <pre><code>Exception in thread "main" org.apache.spark.sql.AnalysisException: Queries with streaming sources must be executed with writeStream.start();;
  kafka
      at org.apache.spark.sql.catalyst.analysis.UnsupportedOperationChecker$.org$apache$spark$sql$catalyst$analysis$UnsupportedOperationChecker$$throwError(UnsupportedOperationChecker.scala:389)
      at org.apache.spark.sql.catalyst.analysis.UnsupportedOperationChecker$$anonfun$checkForBatch$1.apply(UnsupportedOperationChecker.scala:38)
      at org.apache.spark.sql.catalyst.analysis.UnsupportedOperationChecker$$anonfun$checkForBatch$1.apply(UnsupportedOperationChecker.scala:36)
      at org.apache.spark.sql.catalyst.trees.TreeNode.foreachUp(TreeNode.scala:127)
      at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$foreachUp$1.apply(TreeNode.scala:126)
      at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$foreachUp$1.apply(TreeNode.scala:126)
      at scala.collection.immutable.List.foreach(List.scala:392)
      at org.apache.spark.sql.catalyst.trees.TreeNode.foreachUp(TreeNode.scala:126)
      at org.apache.spark.sql.catalyst.analysis.UnsupportedOperationChecker$.checkForBatch(UnsupportedOperationChecker.scala:36)
      at org.apache.spark.sql.execution.QueryExecution.assertSupported(QueryExecution.scala:51)
      at org.apache.spark.sql.execution.QueryExecution.withCachedData$lzycompute(QueryExecution.scala:62)
      at org.apache.spark.sql.execution.QueryExecution.withCachedData(QueryExecution.scala:60)
      at org.apache.spark.sql.execution.QueryExecution.optimizedPlan$lzycompute(QueryExecution.scala:66)
      at org.apache.spark.sql.execution.QueryExecution.optimizedPlan(QueryExecution.scala:66)
      at org.apache.spark.sql.execution.QueryExecution.sparkPlan$lzycompute(QueryExecution.scala:72)
      at org.apache.spark.sql.execution.QueryExecution.sparkPlan(QueryExecution.scala:68)
      at org.apache.spark.sql.execution.QueryExecution.executedPlan$lzycompute(QueryExecution.scala:77)
      at org.apache.spark.sql.execution.QueryExecution.executedPlan(QueryExecution.scala:77)
      at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3360)
      at org.apache.spark.sql.Dataset.collectAsList(Dataset.scala:2794)
      at com.ges.kafka.DatasetKafka.main(DatasetKafka.java:48)
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
  </code></pre>