création RDD et variable de liaison

Question

J'ai un code très simple:

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() 

je ne m'y attendais que la sortie soit la suivante:

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

Cependant, la sortie est la suivante:

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

C'est pour le moins déroutant.

Il semble, en raison de l'évaluation paresseuse de RDD, la valeur de i qui est utilisé pour créer RDD est celui qu'il porte lorsque collect() est appelé, qui est une (de la dernière exécution de la boucle for).

Maintenant, les deux éléments du tuple sont dérivés de i.

Mais il semble, pour le premier élément du tuple, i ours valeurs 0 et 1, tandis que pour le second élément du tuple i porte la valeur 2.

Quelqu'un peut-il expliquer s'il vous plaît ce qui se passe?

Merci.

Answer 1

changer juste

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

à

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

C'est seulement Python, regardez ce

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

Answer 2

sc.parallelize() est une action qui sera exécutée instantanément. Donc, les deux valeurs de i, c'est-à-dire 0 et 1, seront utilisées.

Mais dans le cas de rdd.map(), seule la dernière valeur de i sera utilisée lorsque vous appellerez collect() ultérieurement.

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

Ici rdd.map transform coutume RDD, il suffit de créer DAG (graphe acyclique orienté), la fonction i.e. lambda ne sera pas appliquée aux éléments de RDD.

Lorsque vous appelez Collect(), la fonction lambda sera appelée mais à ce moment i a une valeur de 1. Si vous réaffectez i=10 avant d'appeler recueillir alors que la valeur de i sera utilisée.