Cache and persist

Since each action recomputes the plan from the read (see Spark lazy evaluation), an expensive DataFrame used in several actions is recomputed several times. cache() marks the DataFrame to be kept the first time an action computes it.

base = (spark.table("silver.orders")
        .join(spark.table("silver.customers"), "customer_id")
        .where("status = 'delivered'"))

base.cache()            # only marks it: nothing happens yet
base.count()            # computes and keeps it
base.groupBy("city").count().show()      # reads from the cache
base.groupBy("segment").sum("amount").show()  # reads from the cache
base.unpersist()        # releases it

cache or persist

On a DataFrame, cache() is persist() with the default level, MEMORY_AND_DISK_DESER (whatever does not fit in memory goes to local disk). persist(StorageLevel.DISK_ONLY) and other levels trade memory for speed.

When not to use it

• The DataFrame is used only once: the cache is pure cost (it takes memory and takes space away from execution).
• The read is already cheap and selective (Delta with a file-skipping filter): reading again may cost less than keeping it.
• In a scheduled job, prefer writing the intermediate result as a table: it survives failure, has history and can be read by another job.

• Databricks also has the *disk cache* (formerly called Delta cache), automatic on some node types: it keeps a local copy of the Parquet files read. It is a different thing: it is not the DataFrame's cache().
• Forgot the unpersist()? Spark evicts the oldest when memory runs short, but until then it competes with the real work.