In Pyspark Structured Streaming, how can I discard already generated output before writing to Kafka?

I am trying to do Structured Streaming (Spark 2.4.0) on Kafka source data where I am reading latest data and performing aggregations on a 10 minute window. I am using "update" mode while writing the data.

For example, the data schema is as below:

tx_id, cust_id, product, timestamp

My aim is to find customers who have bought more than 3 products in last 10 minutes. Let's say prod is the dataframe which is read from kafka, then windowed_df is defined as:

windowed_df_1 = prod.groupBy(window("timestamp", "10 minutes"), cust_id).count()
windowed_df = windowed_df_1.filter(col("count")>=3)

Then I am joining this with a master dataframe from hive table "customer_master" to get cust_name:

final_df = windowed_df.join(customer_master, "cust_id")

And finally, write this dataframe to Kafka sink (or console for simplicity)

query = final_df.writeStream.outputMode("update").format("console").option("truncate",False).trigger(processingTime='2 minutes').start()
query.awaitTermination()

Now, when this code runs every 2 minutes, in the subsequent runs, I want to discard all those customers who were already part of my output. I don't want them in my output even if they buy any product again.

Can I write the stream output temporarily somewhere (may be a hive table) and do an "anti-join" for each execution ? This way I can also have a history maintained in a hive table.

I also read somewhere that we can write the output to a memory sink and then use df.write to save it in HDFS/Hive. But what if we terminate the job and re-run ? The in-memory table will be lost in this case I suppose.

Please help as I am new to Structured Streaming.

**

• Update: -

** I also tried below code to write output in Hive table as well as Console(or Kafka sink):

def write_to_hive(df, epoch_id):
    df.persist()
    df.write.format("hive").mode("append").saveAsTable("hive_tab_name")
    pass

final_df.writeStream.outputMode("update").format("console").option("truncate", False).start()

final_df.writeStream.outputMode("update").foreachBatch(write_to_hive).start()

But this only performs the 1st action, i.e. write to Console. If i write "foreachBatch" first, it will save to Hive table but does not print to console.

I want to write to 2 different sinks. Please help.