Spark caching effect on Optimized Logical Plan

user_6933993 0
Viewed 284

I was looking at this question and excellent answer Spark: Explicit caching can interfere with Catalyst optimizer's ability to optimize some queries?

Gist is that this:

val df = spark.range(100)
df.join(df, Seq("id")).filter('id <20).explain(true)

generates a sufficiently robust plan for a system not using indexes by applying filtering first:

== Optimized Logical Plan ==
Project [id#16L]
+- Join Inner, (id#16L = id#18L)
   :- Filter (id#16L < 20)
   :  +- Range (0, 100, step=1, splits=Some(8))
   +- Filter (id#18L < 20)
      +- Range (0, 100, step=1, splits=Some(8))

The example then shows that:

df.join(df, Seq("id")).cache.filter('id <20).explain(true)

generates this plan:

== Optimized Logical Plan ==
Filter (id#16L < 20)
+- InMemoryRelation [id#16L], StorageLevel(disk, memory, deserialized, 1 replicas)
      +- *(2) Project [id#16L]
         +- *(2) BroadcastHashJoin [id#16L], [id#21L], Inner, BuildRight
            :- *(2) Range (0, 100, step=1, splits=8)
            +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, false])), [id=#112]
               +- *(1) Range (0, 100, step=1, splits=8)

Then what about this?

df.join(df, Seq("id")).filter('id <20).cache.explain(true)

that generates:

== Optimized Logical Plan ==
InMemoryRelation [id#16L], StorageLevel(disk, memory, deserialized, 1 replicas)
   +- *(1) Filter (id#16L < 20)
      +- *(1) InMemoryTableScan [id#16L], [(id#16L < 20)]
            +- InMemoryRelation [id#16L], StorageLevel(disk, memory, deserialized, 1 replicas)
                  +- *(2) Project [id#16L]
                     +- *(2) BroadcastHashJoin [id#16L], [id#21L], Inner, BuildRight
                        :- *(2) Range (0, 100, step=1, splits=8)
                        +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, false])), [id=#112]
                           +- *(1) Range (0, 100, step=1, splits=8)

Looking for a clarification.

  • I would have thought that the 1st Opt Log Pl would be gotten with caching as last aspect. Must be simple I suspect. Is it the same thing? I think not.
apache-spark caching
1 Answers

Here I do think you run into a bug in the experiment.

If you run the following in a new spark-shell:

val df = spark.range(100)
df.join(df, Seq("id")).filter('id <20).cache.explain(true)

You will have the following Optimized Logical Plan:

== Optimized Logical Plan ==
InMemoryRelation [id#0L], StorageLevel(disk, memory, deserialized, 1 replicas)
   +- *(2) Project [id#0L]
      +- *(2) BroadcastHashJoin [id#0L], [id#2L], Inner, BuildRight
         :- *(2) Filter (id#0L < 20)
         :  +- *(2) Range (0, 100, step=1, splits=12)
         +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, false]))
            +- *(1) Filter (id#2L < 20)
               +- *(1) Range (0, 100, step=1, splits=12)

with the pushdown predicate correctly pushing filter.

However, in a new spark-shell, if you run:

val df = spark.range(100)
df.join(df, Seq("id")).cache.filter('id <20).explain(true)
df.join(df, Seq("id")).filter('id <20).cache.explain(true)

You will have the following Optimized Logical Plan:

== Optimized Logical Plan ==
InMemoryRelation [id#0L], StorageLevel(disk, memory, deserialized, 1 replicas)
   +- *(1) Filter (id#0L < 20)
      +- *(1) InMemoryTableScan [id#0L], [(id#0L < 20)]
            +- InMemoryRelation [id#0L], StorageLevel(disk, memory, deserialized, 1 replicas)
                  +- *(2) Project [id#0L]
                     +- *(2) BroadcastHashJoin [id#0L], [id#2L], Inner, BuildRight
                        :- *(2) Range (0, 100, step=1, splits=12)
                        +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, false]))
                           +- *(1) Range (0, 100, step=1, splits=12)

with a non optimized plan.

Why ?

This is because we already cached the DAG: df.join(df, Seq("id")).

So even if we write it again with a filter and a cache after the filter, the Spark-Engine will see the join DAG and run it from here, so adding a filter afterwise. For the Spark engine, it is faster to use a cached Dataframe than recomputing the whole DAG.

How to solve ?

One can simply unpersist the dag: df.join(df, Seq("id")).unpersist() and then df.join(df, Seq("id")).filter('id <20).cache.explain(true) gives the right OLP

user_72761070
Related