Spark - isEmpty() trap in Spark

我会带着你远行 2022-02-26 04:59 212阅读 0赞

Versions: Spark 2.1.0

In general Spark's actions reflects logic implemented in a lot of equivalent methods in programming languages. As an example we can consider isEmpty() that in Spark checks the existence of only 1 element and similarly in Java's List. But it can often lead to troubles, especially when more than 1 action is invoked.

This post focuses on Spark's isEmpty() action. The first part explains what happens if it's called. It's illustrated with adequate test case. The second part, still through a test case, shows what can go wrong if isEmpty() is called with other action. The last part gives a tip that could help to prevent problem occurred in the 2nd section of this post.

## isEmpty() internals ##

In other words, when RDD's *isEmpty()* method is called, it checks if RDD has partitions and if there are no entries on them. It's visible in method's implementation that in Spark 2.1.0 looks like:

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 </td> 
 <td> <code>def</code> <code>isEmpty()</code><code>:</code> <code>Boolean </code><code>=</code> <code>withScope { </code> <code>&nbsp;&nbsp;</code><code>partitions.length </code><code>==</code> <code>0</code> <code>|| take(</code><code>1</code><code>).length </code><code>==</code> <code>0</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

The left side of comparison is straightforward since it checks only a property of partitions array. The right part does more operations. It invokes *take(Int)* method. As the name indicates, it takes the first *n* elements encountered in all partitions. Below test case shows the way of working of isEmpty():

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 </td> 
 <td> <code>"isEmpty() method"</code> <code>should </code><code>"invoke the transformation on 1 element"</code> <code>in { </code> <code>&nbsp;&nbsp;</code><code>val</code> <code>numbersRdd </code><code>=</code> <code>sparkContext.parallelize(Seq(</code><code>1</code><code>, </code><code>2</code><code>, </code><code>3</code><code>, </code><code>4</code><code>, </code><code>5</code><code>))</code> <code>&nbsp;&nbsp;</code><code>val</code> <code>numbersAccumulator </code><code>=</code> <code>sparkContext.collectionAccumulator[Int](</code><code>"numbers accumulator"</code><code>)</code> &nbsp; <code>&nbsp;&nbsp;</code><code>val</code> <code>numbersMultipliedBy</code><code>2</code> <code>=</code> <code>numbersRdd.map(number </code><code>=</code><code>&gt; { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>println(s</code><code>"Mapping number ${number}"</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>numbersAccumulator.add(number)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>number * </code><code>2</code> <code>&nbsp;&nbsp;</code><code>})</code> &nbsp; &nbsp; <code>&nbsp;&nbsp;</code><code>if</code> <code>(numbersMultipliedBy</code><code>2</code><code>.isEmpty()) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>println(</code><code>"Numbers are empty"</code><code>)</code> <code>&nbsp;&nbsp;</code><code>}</code> &nbsp; <code>&nbsp;&nbsp;</code><code>numbersAccumulator.value.size shouldEqual(</code><code>1</code><code>)</code> <code>&nbsp;&nbsp;</code><code>numbersAccumulator.value should contain only(</code><code>1</code><code>)</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

## Problematic isEmpty() ##

The nature of isEmpty() should not introduce a lot of problems during processing. After all, it operates only on the 1st encountered element and making 2 actions, the 2nd conditioned by emptiness of RDD, could not be difficult. Below example shows the problematic case:

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 </td> 
 <td> <code>"isEmpty() method invoked with other action"</code> <code>should </code><code>"make accumulated data false"</code> <code>in { </code> <code>&nbsp;&nbsp;</code><code>val</code> <code>numbersRdd </code><code>=</code> <code>sparkContext.parallelize(Seq(</code><code>1</code><code>, </code><code>2</code><code>, </code><code>3</code><code>, </code><code>4</code><code>, </code><code>5</code><code>))</code> <code>&nbsp;&nbsp;</code><code>val</code> <code>numbersAccumulator </code><code>=</code> <code>sparkContext.collectionAccumulator[Int](</code><code>"numbers accumulator"</code><code>)</code> &nbsp; <code>&nbsp;&nbsp;</code><code>val</code> <code>numbersMultipliedBy</code><code>2</code> <code>=</code> <code>numbersRdd.map(number </code><code>=</code><code>&gt; { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>println(s</code><code>"Mapping number ${number}"</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>numbersAccumulator.add(number)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>number * </code><code>2</code> <code>&nbsp;&nbsp;</code><code>})</code> &nbsp; <code>&nbsp;&nbsp;</code><code>// &lt;!----&gt; It's an anti-pattern, written here</code> <code>&nbsp;&nbsp;</code><code>// only to show possible errors of incorrect </code> <code>&nbsp;&nbsp;</code><code>// implementation</code> <code>&nbsp;&nbsp;</code><code>if</code> <code>(!numbersMultipliedBy</code><code>2</code><code>.isEmpty()) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>numbersMultipliedBy</code><code>2</code><code>.saveAsTextFile(outputDir)</code> <code>&nbsp;&nbsp;</code><code>}</code> &nbsp; <code>&nbsp;&nbsp;</code><code>Directory(outputDir).exists should be(</code><code>true</code><code>)</code> <code>&nbsp;&nbsp;</code><code>numbersAccumulator.value.size shouldEqual(</code><code>6</code><code>)</code> <code>&nbsp;&nbsp;</code><code>// the "1" is duplicated because if was read twice:</code> <code>&nbsp;&nbsp;</code><code>// by isEmpty() and by saveAsTextFile() actions</code> <code>&nbsp;&nbsp;</code><code>numbersAccumulator.value should contain theSameElementsAs(Seq(</code><code>1</code><code>, </code><code>1</code><code>, </code><code>2</code><code>, </code><code>3</code><code>, </code><code>4</code><code>, </code><code>5</code><code>))</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

As you can see, the action of saving file is conditioned (for whatever reason) by the emptiness of mapped RDD. As a result we can see that some of RDD elements were computed twice ('1' is accumulated twice). The other problem with it is that obviously 2 jobs are created, both having all computations from scratch. If instead of parallelizing objects we would read a millions of lines from different files, the processing time would very probably increase a lot. But fortunately, there is a solution.

## Problematic isEmpty() solution ##

One of solutions for this kind of operations where two or more actions apply on the same or almost the same data is the use of **cache**. The simplest way of using cache consists on simply calling *cache()* method that will store computed RDDs in executors memory. It's shown in below code:

[?][Link 1]

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 </td> 
 <td> <code>"isEmpty() with other action"</code> <code>should </code><code>"not execute accumulator twice"</code> <code>in { </code> <code>&nbsp;&nbsp;</code><code>val</code> <code>numbersRdd </code><code>=</code> <code>sparkContext.parallelize[Int](Seq(</code><code>1</code><code>, </code><code>2</code><code>, </code><code>3</code><code>, </code><code>4</code><code>, </code><code>5</code><code>))</code> <code>&nbsp;&nbsp;</code><code>val</code> <code>numbersAccumulator </code><code>=</code> <code>sparkContext.collectionAccumulator[Int](</code><code>"numbers accumulator"</code><code>)</code> &nbsp; <code>&nbsp;&nbsp;</code><code>val</code> <code>numbersMultipliedBy</code><code>2</code> <code>=</code> <code>numbersRdd.map(number </code><code>=</code><code>&gt; { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>println(s</code><code>"Mapping number ${number}"</code><code>)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>numbersAccumulator.add(number)</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>number * </code><code>2</code> <code>&nbsp;&nbsp;</code><code>})</code> &nbsp; <code>&nbsp;&nbsp;</code><code>numbersMultipliedBy</code><code>2</code><code>.cache()</code> <code>&nbsp;&nbsp;</code><code>if</code> <code>(!numbersMultipliedBy</code><code>2</code><code>.isEmpty()) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>numbersMultipliedBy</code><code>2</code><code>.saveAsTextFile(outputDir)</code> <code>&nbsp;&nbsp;</code><code>}</code> &nbsp; <code>&nbsp;&nbsp;</code><code>Directory(outputDir).exists should be(</code><code>true</code><code>)</code> <code>&nbsp;&nbsp;</code><code>numbersAccumulator.value.size shouldEqual(</code><code>5</code><code>)</code> <code>&nbsp;&nbsp;</code><code>numbersAccumulator.value should contain theSameElementsAs(Seq(</code><code>1</code><code>, </code><code>2</code><code>, </code><code>3</code><code>, </code><code>4</code><code>, </code><code>5</code><code>))</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

As proven in below test, cache() avoids transformations of the same RDD to be executed the same number of times than the number of declared actions. Caching activity can be detected through logs with following events:

DEBUG Level for block broadcast_0 is StorageLevel(disk, memory, deserialized, 1 replicas) (org.apache.s
    park.storage.BlockManager:58)
    DEBUG Getting local block rdd_1_0 (org.apache.spark.storage.BlockManager:58)
    // ...
    Block rdd_1_0 stored as values in memory (estimated size 40.0 B, free 337.5 MB) (org.apache.spark.storage.memory.MemoryStore:54)
    Added rdd_1_0 in memory on 192.168.0.12:45228 (size: 40.0 B, free: 337.5 MB) (org.apache.spark.storage.BlockManagerInfo:54)

At first glance, isEmpty() action doesn't seem to be invasive. After all, it stops at the 1st found element. It can encourage to use it in with other actions to condition some behavior. The important thing to keep in mind is that isEmpty() triggers real computations. And if these computations are much complex than the ones used in above tests (e.g. by key operations) or if dataset is much bigger, it can slow down the processing. One solution to that could be the use of cache that avoids operations recomputation in the case of two or more actions involved on given RDD.

consult:[https://www.waitingforcode.com/apache-spark/isEmpty-trap-spark/read\#problematic\_isEmpty][https_www.waitingforcode.com_apache-spark_isEmpty-trap-spark_read_problematic_isEmpty]

[Link 1]: https://www.waitingforcode.com/apache-spark/isEmpty-trap-spark/read#
[https_www.waitingforcode.com_apache-spark_isEmpty-trap-spark_read_problematic_isEmpty]: https://www.waitingforcode.com/apache-spark/isEmpty-trap-spark/read#problematic_isEmpty