Comparing two data frames in Spark (performance)

user_7845532 0
Viewed 1612

I need to compare two dataframes in my spark application. I went through the following post. How to obtain the difference between two DataFrames?

However, I don't understand why the approach in the best answer

df1.unionAll(df2).except(df1.intersect(df2))

is better than the one in the question

df1.except(df2).union(df2.except(df1))

Can anyone explain? As per my understanding, the latter works with two smaller datasets and former works with a large dataset. Is it because the latter does a distinct as a part of union? Even then, if it is more likely case that two data frames have same records, we are dealing with a small dataset in the latter case.

apache-spark-sql apache-spark scala java performance
2 Answers

Let's consider a scenario where both df1 and df2 (of size N and M respectively) are too large to be broadcasted, but there is no overlap between df1 and df2.

Let's call it the result di. In such case df1.intersect(df2) will require a full shuffle of N + M rows, however the size of the output will be equal to 0. In such case df1.unionAll(df2).except(di) can be executed as a broadcast join (such optimization might require adaptive execution unless specific plan is forced by the user). It is also important to note that such plan doesn't require caching.

In contrast the cost of df1.except(df2).union(df2.except(df1)) will be constant in regards to the cardinality of the intersection.

At the same time, if d1 is to large to be broadcasted, it already has a partitioning compatible with except, so the remaining query shouldn't require additional shuffle.

user_104653550

First things first - unionAll is deprecated as of version 2 of Spark. Please use union instead, as you did in the second snippet.

Second, in the answers to the question, you're referencing to, there is no information, that the first piece of code is better. I have prepared such a scenario. For me, the first one took 31s and the second 18s. In my case df1 has ~3 million rows and df2 ~ 1 million, 5 columns each.

If we now analyse the optimized logical execution plan for the first query:

== Optimized Logical Plan ==
GlobalLimit 21
+- LocalLimit 21
   +- Aggregate [_c0#10, _c1#11, _c2#12, _c3#13, _c4#14], [cast(_c0#10 as string) AS _c0#67, cast(_c1#11 as string) AS _c1#68, cast(_c2#12 as string) AS _c2#69, cast(_c3#13 as string) AS _c3#70, cast(_c4#14 as string) AS _c4#71]
      +- Join LeftAnti, (((((_c0#10 <=> _c0#52) && (_c1#11 <=> _c1#53)) && (_c2#12 <=> _c2#54)) && (_c3#13 <=> _c3#55)) && (_c4#14 <=> _c4#56))
         :- Union
         :  :- Relation[_c0#10,_c1#11,_c2#12,_c3#13,_c4#14] csv
         :  +- Project [_c0#30, _c1#31, _c2#32, _c3#33, cast(_c4#34 as double) AS _c4#40]
         :     +- Relation[_c0#30,_c1#31,_c2#32,_c3#33,_c4#34] csv
         +- Aggregate [_c0#52, _c1#53, _c2#54, _c3#55, _c4#56], [_c0#52, _c1#53, _c2#54, _c3#55, _c4#56]
            +- Join LeftSemi, (((((_c0#52 <=> _c0#30) && (_c1#53 <=> _c1#31)) && (_c2#54 <=> _c2#32)) && (_c3#55 <=> _c3#33)) && (_c4#56 <=> _c4#46))
               :- Relation[_c0#52,_c1#53,_c2#54,_c3#55,_c4#56] csv
               +- Project [_c0#30, _c1#31, _c2#32, _c3#33, cast(_c4#34 as double) AS _c4#46]
                  +- Relation[_c0#30,_c1#31,_c2#32,_c3#33,_c4#34] csv

We can see, that there is Union and Join(intersection) running concurrently, which is very costly, especially Union, whereas for the second query:

== Optimized Logical Plan ==
GlobalLimit 21
+- LocalLimit 21
   +- Union
      :- LocalLimit 21
      :  +- Aggregate [_c0#10, _c1#11, _c2#12, _c3#13, _c4#14], [cast(_c0#10 as string) AS _c0#120, cast(_c1#11 as string) AS _c1#121, cast(_c2#12 as string) AS _c2#122, cast(_c3#13 as string) AS _c3#123, cast(_c4#14 as string) AS _c4#124]
      :     +- Join LeftAnti, (((((_c0#10 <=> _c0#30) && (_c1#11 <=> _c1#31)) && (_c2#12 <=> _c2#32)) && (_c3#13 <=> _c3#33)) && (_c4#14 <=> _c4#98))
      :        :- Relation[_c0#10,_c1#11,_c2#12,_c3#13,_c4#14] csv
      :        +- Project [_c0#30, _c1#31, _c2#32, _c3#33, cast(_c4#34 as double) AS _c4#98]
      :           +- Relation[_c0#30,_c1#31,_c2#32,_c3#33,_c4#34] csv
      +- LocalLimit 21
         +- Aggregate [_c0#30, _c1#31, _c2#32, _c3#33, _c4#104], [cast(_c0#30 as string) AS _c0#130, cast(_c1#31 as string) AS _c1#131, cast(_c2#32 as string) AS _c2#132, cast(_c3#33 as string) AS _c3#133, cast(_c4#104 as string) AS _c4#134]
            +- Join LeftAnti, (((((_c0#30 <=> _c0#10) && (_c1#31 <=> _c1#11)) && (_c2#32 <=> _c2#12)) && (_c3#33 <=> _c3#13)) && (_c4#104 <=> _c4#14))
               :- Project [_c0#30, _c1#31, _c2#32, _c3#33, cast(_c4#34 as double) AS _c4#104]
               :  +- Relation[_c0#30,_c1#31,_c2#32,_c3#33,_c4#34] csv
               +- Relation[_c0#10,_c1#11,_c2#12,_c3#13,_c4#14] csv

There are two LeftAnti running concurrently (relative compliments). This takes less space and is more efficient. This can be seen in SparkUI:

First query: First query Second query: Second query

In the first case, stage 7 - Union is the most costly, whereas in the second case stages 42 and 41 (above) are relatively faster.

user_76067640
Related