wrong key in pyspark while using flatMapValues

user_16162063 0
Viewed 38

I want to add the 'key' which is used in flatMapValues, but i always get the wrong one.

Here is rdd.collect()

[{'a': 1, 'b': 2, 'c': [1, 2, 3, 4]},
 {'a': 11, 'b': 22, 'c': [5, 6, 7, 8]},
 {'a': 11, 'b': 23, 'c': [5, 6, 7, 8]}]

and the operation is

def add_key(x):
     x[0]['key'] = x[1]
     return x
rdd.map(lambda x: (x, x['c'])).flatMapValues(lambda x: x).map(add_key).map(lambda x:(x[1],x[0])

but I got the result like

[(1, {'a': 1, 'b': 2, 'c': [1, 2, 3, 4], 'key': 1}),
 (2, {'a': 1, 'b': 2, 'c': [1, 2, 3, 4], 'key': 3}),
 (3, {'a': 1, 'b': 2, 'c': [1, 2, 3, 4], 'key': 3}),
 (4, {'a': 1, 'b': 2, 'c': [1, 2, 3, 4], 'key': 4}),
 (5, {'a': 11, 'b': 22, 'c': [5, 6, 7, 8], 'key': 5}),
 (6, {'a': 11, 'b': 22, 'c': [5, 6, 7, 8], 'key': 7}),
 (7, {'a': 11, 'b': 22, 'c': [5, 6, 7, 8], 'key': 7}),
 (8, {'a': 11, 'b': 22, 'c': [5, 6, 7, 8], 'key': 8}),
 (5, {'a': 11, 'b': 23, 'c': [5, 6, 7, 8], 'key': 5}),
 (6, {'a': 11, 'b': 23, 'c': [5, 6, 7, 8], 'key': 7}),
 (7, {'a': 11, 'b': 23, 'c': [5, 6, 7, 8], 'key': 7}),
 (8, {'a': 11, 'b': 23, 'c': [5, 6, 7, 8], 'key': 8})]

where the 'key' is partially wrong, the result should be

[(1, {'a': 1, 'b': 2, 'c': [1, 2, 3, 4], 'key': 1}),
 (2, {'a': 1, 'b': 2, 'c': [1, 2, 3, 4], 'key': 2}),
 (3, {'a': 1, 'b': 2, 'c': [1, 2, 3, 4], 'key': 3}),
 (4, {'a': 1, 'b': 2, 'c': [1, 2, 3, 4], 'key': 4}),
 (5, {'a': 11, 'b': 22, 'c': [5, 6, 7, 8], 'key': 5}),
 (6, {'a': 11, 'b': 22, 'c': [5, 6, 7, 8], 'key': 6}),
 (7, {'a': 11, 'b': 22, 'c': [5, 6, 7, 8], 'key': 7}),
 (8, {'a': 11, 'b': 22, 'c': [5, 6, 7, 8], 'key': 8}),
 (5, {'a': 11, 'b': 23, 'c': [5, 6, 7, 8], 'key': 5}),
 (6, {'a': 11, 'b': 23, 'c': [5, 6, 7, 8], 'key': 6}),
 (7, {'a': 11, 'b': 23, 'c': [5, 6, 7, 8], 'key': 7}),
 (8, {'a': 11, 'b': 23, 'c': [5, 6, 7, 8], 'key': 8})]

1

rdd pyspark apache-spark python
1 Answers

It's an interesting (bug?) that I do not understand. But here is an alternative that should give the expected results:

rdd2 = (rdd
    .map(lambda x: (x, x['c']))
    .flatMapValues(lambda x: x)
    .map(lambda x: (x[1], {**x[0], **{'key':x[1]}}))
)

rdd2.collect()
[(1, {'a': 1, 'b': 2, 'c': [1, 2, 3, 4], 'key': 1}),
 (2, {'a': 1, 'b': 2, 'c': [1, 2, 3, 4], 'key': 2}),
 (3, {'a': 1, 'b': 2, 'c': [1, 2, 3, 4], 'key': 3}),
 (4, {'a': 1, 'b': 2, 'c': [1, 2, 3, 4], 'key': 4}),
 (5, {'a': 11, 'b': 22, 'c': [5, 6, 7, 8], 'key': 5}),
 (6, {'a': 11, 'b': 22, 'c': [5, 6, 7, 8], 'key': 6}),
 (7, {'a': 11, 'b': 22, 'c': [5, 6, 7, 8], 'key': 7}),
 (8, {'a': 11, 'b': 22, 'c': [5, 6, 7, 8], 'key': 8}),
 (5, {'a': 11, 'b': 23, 'c': [5, 6, 7, 8], 'key': 5}),
 (6, {'a': 11, 'b': 23, 'c': [5, 6, 7, 8], 'key': 6}),
 (7, {'a': 11, 'b': 23, 'c': [5, 6, 7, 8], 'key': 7}),
 (8, {'a': 11, 'b': 23, 'c': [5, 6, 7, 8], 'key': 8})]
user_141657300
Related