Объедините несколько dict в другой dict в pyspark rdd - PullRequest
Новая
       13

Объедините несколько dict в другой dict в pyspark rdd

1 голос
/ 19 июня 2019

У меня есть следующий фрейм данных: 

from pyspark.sql import SparkSession

sqlContext = SparkSession.builder.appName("test").enableHiveSupport().getOrCreate()
data = [(1,2,0.1,0.3),(1,2,0.1,0.3),(1,3,0.1,0.3),(1,3,0.1,0.3),
        (11, 12, 0.1, 0.3),(11,12,0.1,0.3),(11,13,0.1,0.3),(11,13,0.1,0.3)]

trajectory_df = sqlContext.createDataFrame(data, schema=['grid_id','rider_id','lng','lat'])
trajectory_df.show()

+-------+--------+---+---+
|grid_id|rider_id|lng|lat|
+-------+--------+---+---+
|      1|       2|0.1|0.3|
|      1|       2|0.1|0.3|
|      1|       3|0.1|0.3|
|      1|       3|0.1|0.3|
|     11|      12|0.1|0.3|
|     11|      12|0.1|0.3|
|     11|      13|0.1|0.3|
|     11|      13|0.1|0.3|
+-------+--------+---+---+

Я хочу объединить данные из той же сетки в DICT. Где rider_id - ключ dict, а широта и долгота - значение dict.

Результаты, которые я ожидаю, следующие: 

[(1, {3:[[0.1, 0.3], [0.1, 0.3]],2:[[0.1, 0.3], [0.1, 0.3]]}),
 (11,{13:[[0.1, 0.3], [0.1, 0.3]],12:[[0.1, 0.3], [0.1, 0.3]]})]

Я могу использовать groupByKey() для группировки grid_id. 

def trans_point(row):
    return ((row.grid_id, row.rider_id), [row.lng, row.lat])
trajectory_df = trajectory_df.rdd.map(trans_point).groupByKey().mapValues(list)
print(trajectory_df.take(10))

[((1, 3), [[0.1, 0.3], [0.1, 0.3]]), ((11, 13), [[0.1, 0.3], [0.1, 0.3]]), ((1, 2), [[0.1, 0.3], [0.1, 0.3]]), ((11, 12), [[0.1, 0.3], [0.1, 0.3]])]

Но я не могу получить результат, когда объединяю несколько dict: 

trajectory_df = trajectory_df.map(lambda x:(x[0][0],{x[0][1]:x[1]})).reduceByKey(lambda x,y:x.update(y))
print(trajectory_df.take(10))
[(1, None), (11, None)]

Я надеюсь, что это сделано по типу RDD по некоторым причинам. Как мне этого добиться? Заранее спасибо.

1 Ответ

1 голос
/ 19 июня 2019

dict.update работает на месте и возвращает None.Из документов:

Обновите словарь парами ключ / значение из других, перезаписав существующие ключи.Возврат None.

Вам необходимо написать собственную функцию приведения, чтобы объединить словари.Мы можем позаимствовать у @ Аарона Холла ответ на Как объединить два словаря в одном выражении? 

def merge_two_dicts(x, y):
    """From https://stackoverflow.com/a/26853961/5858851"""
    z = x.copy()   # start with x's keys and values
    z.update(y)    # modifies z with y's keys and values & returns None
    return z

trajectory_df = trajectory_df.map(lambda x:(x[0][0],{x[0][1]:x[1]}))\
    .reduceByKey(merge_two_dicts)

print(trajectory_df.collect())
#[(1, {2: [[0.1, 0.3], [0.1, 0.3]], 3: [[0.1, 0.3], [0.1, 0.3]]}),
# (11, {12: [[0.1, 0.3], [0.1, 0.3]], 13: [[0.1, 0.3], [0.1, 0.3]]})]
...