создание РДД и переменной связывания

Question

У меня есть очень простой код:

def fun(x, n): 
    return (x, n) 

rdds = [] 
for i in range(2): 
    rdd = sc.parallelize(range(5*i, 5*(i+1))) 
    rdd = rdd.map(lambda x: fun(x, i)) 
    rdds.append(rdd) 

a = sc.union(rdds) 
print a.collect() 

я ожидал выход быть следующее:

[(0, 0), (1, 0), (2, 0), (3, 0), (4, 0), (5, 1), (6, 1), (7, 1), (8, 1), (9, 1)] 

Однако вывод состоит в следующем:

[(0, 1), (1, 1), (2, 1), (3, 1), (4, 1), (5, 1), (6, 1), (7, 1), (8, 1), (9, 1)] 

Это смущает, если не сказать больше.

Кажется, из-за ленивые оценки РДА, значение i, который используется для создания РДА является тот, когда он носит collect() называется, что 1 (от последнего запуска цикла for).

Теперь оба элемента кортежа состоят из i.

Но, кажется, для первого элемента кортежа, i медведей значения 0 и 1, а для второго элемента кортежа i имеет значение 2.

Может кто-нибудь объяснить, что происходит?

Спасибо.

Answer 1

просто изменить

rdd = rdd.map(lambda x: fun(x, i)) 

в

rdd = rdd.map(lambda x, i=i: (x, i)) 

То есть только Python, посмотрите на эту

https://docs.python.org/2.7/tutorial/controlflow.html#default-argument-values

Answer 2

sc.parallelize() - это действие, которое будет выполнено мгновенно. Таким образом, будут использоваться значения i i.e 0 и 1.

Но в случае rdd.map() только последнее значение i будет использовано, когда вы позвоните по телефону collect().

rdd = sc.parallelize(range(5*i, 5*(i+1))) 
rdd = rdd.map(lambda x: fun(x, i)) 

Здесь rdd.map обыкновение превращать ДРР, это будет просто создать DAG (ориентированный ациклический граф), т.е. лямбда-функция не будет применяться к элементам РДУ.

При вызове сбора(), то лямбда-функция будет вызываться, но к тому времени i имеет значение 1. Если вы переназначить i=10 перед вызовом собирать то будет использоваться это значение i.