Как преобразовать фрейм данных Pyspark в словарь Python

Question

У меня новичок в pyspark, есть требование, подобное приведенному ниже

Фрейм данных, имеющий два столбца с (id и data_list) с сортировкой data_list по группам, как показано ниже

+---+-----+-----+
| id| data|value|
+---+-----+-----+
|1_a|AB,Ca|   10|
|1_a|Cd,da|    5|
|1_a|aC,BE|   15|
|1_a|ER,rK|   20|
|2_b|JK,Lh| 1500|
|2_b|Yu,HK|  500|
|2_b|MK,HN|  100|
+---+-----+-----+

afterотсортированный список данных

+---+--------------------+
| id|           data_list|
+---+--------------------+
|1_a|[Cd,da, AB,Ca, aC...|
|2_b|[MK,HN, Yu,HK, JK...|
+---+--------------------+

применение преобразования карты на DF для получения желаемого результата (словарь списка Python),

data = order_df.rdd.map(lambda (x, y): (x.split("_")[1].lower(), (x.split("_")[0].lower(), y))) \
    .groupByKey().mapValues(list)

output

  [('b', [('2', '[MK,HN, Yu,HK, JK,Lh]')]), ('a', [('1', '[Cd,da, AB,Ca, aC,BE, ER,rK]')])] 

, затем итерация спискачтобы получить каждый элемент, как показано ниже

for dd in data.collect():
    print "==========", dd[1][0][1]
    for r in dd[1][0][1]:
        print r + "---"

Желаемый вывод

Cd,da
AB,Ca
aC,BE
ER,rK
....

, но получим как показано ниже

   ========== [Cd,da, AB,Ca, aC,BE, ER,rK]
ttttt:  <type 'str'>
[
C
d
,
d
a
,

A
B
,
C
a
,

a
C
,
B
E
,

E
R
,
r
K
]

Ниже приведен код, пытающийся получить вывод.

 from pyspark import SparkContext, SparkConf
        from pyspark.sql import SQLContext
        from pyspark.sql.types import *
        from pyspark.sql import functions as F
        import operator

        conf = SparkConf().setMaster("local").setAppName("Demo DF")
        sc = SparkContext(conf=conf)
        sqlContext = SQLContext(sparkContext=sc)
        sqlContext.setConf("spark.sql.shuffle.partitions", "3")

       def foo((x, y)):
          z = x.lower().split('_')
          return (z[1], (z[0], ast.literal_eval(json.dumps(y, 
               ensure_ascii=False).encode('utf8'))))

        # define udf
        def sorter(l):
            res = sorted(l, key=operator.itemgetter(1))
            return [item[0] for item in res]


        sort_udf = F.udf(sorter)

       ll_list = [("1_a", "AB,Ca", 10), ("1_a", "Cd,da", 5), ("1_a", "aC,BE", 15), ("1_a", "ER,rK", 20),
               ("2_b", "JK,Lh", 1500), ("2_b", "Yu,HK", 500), ("2_b", "MK,HN", 100)]
    input_df = sc.parallelize(ll_list).toDF(["id", "data", "value"])
        input_df.show()

        # create list column
        grouped_df = input_df.groupby("id") \
            .agg(F.collect_list(F.struct("data", "value")) \
                 .alias("list_col"))

        # test
        order_df = grouped_df.select("id", sort_udf("list_col") \
                                     .alias("data_list"))
        order_df.show()
        data = order_df.rdd.map(foo).groupByKey().mapValues(list)

        for dd in data.collect():
            print "==========", dd[1][0][1]
            for r in dd[1][0][1]:
                print r + "---"

Не могли бы вы помочь мне с этим кодом, чтобы получить правильный вывод.

Answer 1

Проблема в том, что «data_list» на самом деле является столбцом строк:

order_df.dtypes
# [('id', 'string'), ('data_list', 'string')]

Вы можете использовать ast.literal_eval для их анализа.

import ast 

def foo((x, y)): 
    z = x.lower().split('_')
    return (z[1], (z[0], ast.literal_eval(y)))

order_df.rdd.map(foo).groupByKey().mapValues(list).collect()