I currently have a pipeline architecture in a spark project with processes (A, B, C, D) where the output of process A is the input to process B, and the output of process B is the input to process C, etc... A process can contain one or more spark sql queries, and outputs a cached dataframe. This is a linear pipeline, so process C will only use the output of process B, and not the output of process A.
As you might imagine, all this caching takes up a lot of memory, especially on large datasets. When on process C, I no longer need the output of process A, just the output from process B. Normally with spark if I uncache the output of process A, then the output of process B also gets uncached, which I need in process C. So to handle this I do the following:
val new_output = spark.sqlContext.createDataFrame(output.rdd, output.schema)
new_output.createOrReplaceTempView("new_output")
spark.sql("CACHE TABLE new_output")
inputDf.unpersist(true)
return new_output
When I do this, it basically cuts the dag, so I can uncache the input dataframes. When I do an explain on the new_output dataframe it looks like this:
== Parsed Logical Plan ==
LogicalRDD [a#168], false
Question: I am trying to see if this approach could cause issues down the line related to resiliency with spark, or in other ways.
I am fully aware I could use spark checkpointing to insert checkpoints into the DAG. With the above approach, I do not have to spend time writing data to disk / external storage, like a checkpoint would. I would like to keep this all in the same spark application and persist the data in memory between processes.
Update1: Added more code to show the order of caching and uncaching. Using eager caching so the new_output dag is executed and stored in memory so the inputDf can be uncached.