title: streaming aggregate not writing into sink
body: <p>I have to process some files which arrive to me daily. The information have primary key (date,client_id,operation_id). So I created a Stream which append only new data into a delta table:</p>
  
  <pre><code>operations\
          .repartition('date')\
          .writeStream\
          .outputMode('append')\
          .trigger(once=True)\
          .option("checkpointLocation", "/mnt/sandbox/operations/_chk")\
          .format('delta')\
          .partitionBy('date')\
          .start('/mnt/sandbox/operations')
  </code></pre>
  
  <p>This is working fine, but i need to summarize this information grouped by (date,client_id), so i created another streaming from this operations table to a new table. So i tried to convert my <code>date</code> field to a timestamp, so i could use append mode while writing an aggregated stream:</p>
  
  <pre><code>import pyspark.sql.functions as F
  
  summarized= spark.readStream.format('delta').load('/mnt/sandbox/operations')
  summarized= summarized.withColumn('timestamp_date',F.to_timestamp('date'))
  summarized= summarized.withWatermark('timestamp_date','1 second').groupBy('client_id','date','timestamp_date').agg(&lt;lot of aggs&gt;)
  
  summarized\
          .repartition('date')\
          .writeStream\
          .outputMode('append')\
          .option("checkpointLocation", "/mnt/sandbox/summarized/_chk")\
          .trigger(once=True)\
          .format('delta')\
          .partitionBy('date')\
          .start('/mnt/sandbox/summarized')
  </code></pre>
  
  <p>This code runs, but it does not write anything in the sink.</p>
  
  <p>why it isn't writing results into sink?</p>