通过字典将 pyspark 行列表转换为 pandas 数据框答案

【问题标题】：Transform list of pyspark rows into pandas data frame through a dictionary通过字典将 pyspark 行列表转换为 pandas 数据框
【发布时间】：2020-07-07 08:11:02
【问题描述】：

我正在尝试使用字典理解将 PySpark 排序行列表转换为 Pandas 数据框，但仅在明确说明所需字典的键和值时才有效。

row_list = sorted(data, key=lambda row: row['date'])

future_df = {'key': int(key),
             'date': map(lambda row: row["date"], row_list),
             'col1': map(lambda row: row["col1"], row_list),
             'col2': map(lambda row: row["col2"], row_list)}

然后将其转换为 Pandas：

pd.DataFrame(future_df)

此操作可在以下调用的类 ForecastByKey 中找到：

rdd = df.select('*')
    .rdd \
    .map(lambda row: ((row['key']), row)) \
    .groupByKey() \
    .map(lambda args: spark_ops.run(args[0], args[1]))

到目前为止，一切正常；意思是明确指出字典中的列future_df。

当尝试使用以下内容转换整组列 (700+) 时会出现问题：

future_df = {'key': int(key),
             'date': map(lambda row: row["date"], row_list)}

for col_ in columns:
    future_df[col_] = map(lambda row: row[col_], row_list)

pd.DataFrame(future_df)

其中columns 包含传递给ForecastByKey 类的每个字段的名称。

此操作的结果是具有空列或接近零列的数据框。

我正在使用 Python 3.6.10 和 PySpark 2.4.5

如何进行此迭代以获得具有正确信息的数据框？

【问题讨论】：

标签： pandas lambda pyspark group-by apache-spark-sql

【解决方案1】：

经过一番研究，我意识到这可以通过以下方式解决：

row_list = sorted(data, key=lambda row: row['date'])

def f(x):
    return map(lambda row: row[x], row_list)

pre_df = {col_: col_ for col_ in self.sdf_cols}

future_df = toolz.valmap(f, pre_df)

future_df['key'] = int(key)

【讨论】：