Hi @MuthuLakshmi thank you for your response. No I don't use df.cache() anywhere in the code. Yet I tried uncaching intermediate table which is read and updated within the loop, but it didn't help:

spark.catalog.uncacheTable("IM_AWL_Products")

I don't want to disable cluster level caching because we have an entire team running their code in same cluster. So my preference is to solve this more within the code.