title: Error when using collect_list in Spark Structured Streaming
body: <p>I have two versions of Spark code. The first uses structured streaming with a Kafka source:</p>
  
  
  
  <pre class="lang-scala prettyprint-override"><code>dfStream.printSchema()
  //root
  //|-- dt: string (nullable = true)
  //|-- ip: string (nullable = true)
  //|-- device: string (nullable = true)
  
  val dfWindowed = dfStream                                                
      .groupBy($"ip")
      .agg(concat_ws(",", collect_list($"device")).alias("devices"))
      .writeStream                                             
      .outputMode("complete")                                              
      .format("memory")                                                        
      .start()           
  </code></pre>
  
  <p>The second reads from files. But the data is really the same as above:</p>
  
  <pre class="lang-scala prettyprint-override"><code>logDF.printSchema() 
  //root
  //|-- dt: string (nullable = true)
  //|-- ip: string (nullable = true)
  //|-- device: string (nullable = true)
  
  logDF.repartition(32)                                                                                                   
      .groupBy("ip")
      .agg(concat_ws(",", collect_list($"device")).alias("devices")) 
  </code></pre>
  
  <p>The problem is, while the second works well, the first keeps giving me the following error:  </p>
  
  <pre class="lang-none prettyprint-override"><code>at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runBatches(StreamExecution.scala:284)
      at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:177)
  Caused by: org.apache.spark.SparkException: Job aborted due to stage failure: Task 9 in stage 1.0 failed 4 times, most recent failure: Lost task 9.3 in stage 1.0 (TID 28, c3-hadoop-prc-st3417.bj, executor 3): java.lang.RuntimeException: Collect cannot be used in partial aggregations.
  </code></pre>
  
  <p>A long sentence.. but the error seems to be the following: </p>
  
  <pre class="lang-none prettyprint-override"><code>java.lang.RuntimeException: Collect cannot be used in partial aggregations.
  </code></pre>
  
  <p>I found a couple of related SO questions, but no solutions worked so far. Any suggestions on the following are much appreciated: </p>
  
  <ol>
  <li>the meaning of "partial aggregation" &amp; the reason why no such problems with static (non-stream) dataset,</li>
  <li>a workaround...</li>
  </ol>