PySpark Evaluation

Posted on

Question :

PySpark Evaluation

I am trying the following code which adds a number to every row in an RDD and returns a list of RDDs using PySpark.

from pyspark.context import SparkContext
file  = "file:///home/sree/code/scrap/sample.txt"
sc = SparkContext('local', 'TestApp')
data = sc.textFile(file) 
splits = [ p :  int(p) + i) for i in range(4)]
print splits[0].collect()
print splits[1].collect()
print splits[2].collect()

The content in the input file (sample.txt) is:


I was expecting an output like this (adding the numbers in the rdd with 0, 1, 2 respectively):


whereas the actual output was :

[4, 5, 6]
[4, 5, 6]
[4, 5, 6]

which means that the comprehension used only the value 3 for variable i, irrespective of the range(4).

Why does this behavior happen ?

Asked By: srjit


Answer #1:

It happens because of Python late binding and is not (Py)Spark specific. i will be looked-up when lambda p : int(p) + i is used, not when it is defined. Typically it means when it is called but in this particular context it is when it is serialized to be send to the workers.

You can do for example something like this:

def f(i):
    def _f(x):
            return int(x) + i
    return _f

data = sc.parallelize(["1", "2", "3"])
splits = [ for i in range(4)]
[rdd.collect() for rdd in splits]
## [[1, 2, 3], [2, 3, 4], [3, 4, 5], [4, 5, 6]]
Answered By: zero323

Answer #2:

This is due to to the fact that lambdas refer to the i via reference! It has nothing to do with spark.
See this

You can try this:

a =[(lambda y: (lambda x: y + int(x)))(i) for i in range(4)]
splits = [[x]) for x in range(4)]

or in one line

splits = [[(lambda y: (lambda x: y + int(x)))(i) for i in range(4)][x])
    for x in range(4)
Answered By: Himaprasoon

Leave a Reply

Your email address will not be published. Required fields are marked *