title: Spark MapGroupsWithState Update function doesn't allow action queries like filter, count, show etc
body: <p>I have an IoT application where I receive the data from different Energy Meters and Inverter Meters. These meters continuously sends meter value as no of units consumed. This value continuously increments over period of time and I have to calculate One hour Energy Consumption for each meter.</p>
  
  <p>I get all this data in a kafka topic from which I create Structured Streaming Dataframe.  </p>
  
  <p>On this expanded dataframe I am applying mapGroupsWithState function. This should return me one hour generation. </p>
  
  <p>The Problem:
  I am not able to do any of the count, show, filter, aggregation operations inside update function using spark dataframe.</p>
  
  <pre><code>    val df_output = final_df
        .selectExpr("*")
        .as[input_row_druid]
        .groupByKey(_.plant_slug)
        .mapGroupsWithState(GroupStateTimeout.ProcessingTimeTimeout)(updateAcrossEvents)
        .writeStream
        .format("console")
        .outputMode("update")
        .start()
  
      df_output.awaitTermination()
  
  
    def updateAcrossEvents(plant_slug:String, inputs: Iterator[input_row_druid],
                           oldState: GroupState[source_state]):out_state = {
      val spark_session = SparkSession.builder().getOrCreate()
      import spark_session.implicits._
  
      val list_of_list = inputs.toList
      val new_df = list_of_list.toDF
      var my_state:source_state = if (oldState.exists) oldState.get else source_state(plant_slug,list_of_list)
  
      println("Printing inverter and merter df with counts")
      val inverter_df = new_df.filter($"device_type" === "INVERTER")
      val meter_df = new_df.filter($"device_type" === "METER")
      println(inverter_df.show())
      val inv_count = inverter_df.count()
      println(meter_df.show())
      val meter_count = meter_df.count()
      println(inv_count)
      println(meter_count)
  
      val new_state = source_state(plant_slug, list_of_list)
      oldState.update(new_state)
      var out = out_state(plant_slug,inv_count.toString,meter_count.toString)
      out
    }
  </code></pre>
  
  <p>If I do not do any filter operation new_df.show prints all data in dataframe. But count, show, filter not functioning.</p>
  
  <p>One thing I noticed that is whenever I submit job to spark it runs in multiple batches, batch 0 is always successful but it hangs on batch 0. It never proceeds in next batch.</p>
  
  <p>This is expected Result</p>
  
  <pre><code>+--------------------+-------+---------+
  |          plant_slug|inv_gen|meter_gen|
  +--------------------+-------+---------+
  |              plant1|     11|       10|
  |              plant2|     20|       19|
  |              plant3|     40|       38|
  |              plant4|     59|       57|
  |              plant5|     37|       35|
  +--------------------+-------+---------+
  </code></pre>