Pyspark RDD filter behaviour when filter condition variable is modified

user_7362886 0
Viewed 214

When the following code is run:

A = ss.sparkContext.parallelize(range(1, 100))
t = 50
B = A.filter(lambda x: x < t)
print(B.count())
t = 10
C = B.filter(lambda x: x > t)
print(C.count())

The output is:

49
0

Which is incorrect as there are 39 values between 10 and 49. It seems like changing t to 10 from 50 effected the first filter as well and it got re-evalutated so when both filters are applied consecutively it effectively becomes x<10 which would result in 1, 2, 3, 4, 5, 6 ,7, 8, 9 followed by x>10 resulting in an empty rdd.

But When I add debug prints in the code the result is not what I expect and I am looking for an explanation:

A = ss.sparkContext.parallelize(range(1, 100))
t = 50
B = A.filter(lambda x: x < t)
print(B.collect())
t = 10
print(B.collect())
print(B.count())
C = B.filter(lambda x: x > t)
print(C.collect())
print(C.count())

The output is:

[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49]
[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49]
9
[]
0

How come the count is 10 after t=10 but print(B.collect()) shows the expected rdd with values from 1 to 49? If triggering collect after changing t re-did the filter, then shouldn't collect() show values from 1-9?

I am new to pyspark, I suspect this has to do with spark's lazy operations and caching. Can someone explain what is going on behind the scenes?

Thanks!

python-3.9 rdd pyspark filter
1 Answers

Your assumption is correct, the observed behaviour is related to Spark's lazy evaluation of transformations.

When B.count() is executed, Spark simply applies the filter x < t with t = 50 and prints the expected value of 49.

When C.count() is executed, Spark sees two filters in the excution plan of C, namely x < t and x > t. At this point in time t has been set to 10 and no element of the rdd satiesfies both conditions to be smaller and larger than 10. Spark ignores that fact that the first filter has already been evaluated. When a Spark action is called all transformations in the history of the current rdd are executed (unless some intermediate result has been cached, see below).

A way to examine this behaviour (a bit) more in details is to switch to Scala and print toDebugString for both rdds.1

println(B.toDebugString)

prints

(4) MapPartitionsRDD[1] at filter at SparkStarter.scala:23 []
 |  ParallelCollectionRDD[0] at parallelize at SparkStarter.scala:19 []

while

println(C.toDebugString)

prints

(4) MapPartitionsRDD[2] at filter at SparkStarter.scala:28 []
 |  MapPartitionsRDD[1] at filter at SparkStarter.scala:23 []
 |  ParallelCollectionRDD[0] at parallelize at SparkStarter.scala:19 []

Here we can see that for rdd B one filter is applied and for rdd C two filters are applied.

How to fix the issue?

If the result of the first filter is cached the expected result is printed out. When then t is changed and the second filter is applied C.count() only triggers the second filter based on the cached result of B:

A = ss.sparkContext.parallelize(range(1, 100))
t = 50
B = A.filter(lambda x: x < t).cache()
print(B.count())
t = 10
C = B.filter(lambda x: x > t)
print(C.count())

prints the expected result.

49
39

1 Unfortunately this works only in the Scala version of Spark. PySpark seems to "condense" the output of toDebugString (version 3.1.1).

user_21298010
Related