Spark：数据框转换答案

【问题标题】：Spark: Dataframe TransformationSpark：数据框转换
【发布时间】：2021-08-13 07:47:28
【问题描述】：

我有 dataframe_1：

+-------------+----+---------+
|         Name| Age|   Salary|
+-------------+----+---------+
|Avery Bradley|25.0|7730337.0|
|  Jae Crowder|25.0|6796117.0|
+-------------+----+---------+

并希望将其转换为 dataframe_2：

+----------------------------------------------------------------------------------------------------------------------+
|         json_data                                                                                                    |
+----------------------------------------------------------------------------------------------------------------------+
|[{"Name": "Avery Bradley", "Age": 25.0, "Salary" 7730337.0}, {"Name": "Jae Crowder", "Age": 25.0, "Salary" 6796117.0}]|    
+----------------------------------------------------------------------------------------------------------------------+

我可以做dataframe_1.toPandas().to_dict(orient="records")，但这将是 dataframe 到 dict(json object) 的转换，我需要 dataframe 到 dataframe 的转换。

如果可能的话，我们将不胜感激 PySpark 中的解决方案。

【问题讨论】：

this 可能会有所帮助。它使用to_json() 函数。
嗨@samkart，你提到的答案为每一行创建了json，我需要将所有行合并为一行，格式为list of json
使用collect_list() 的帖子——基于上述内容，而不是在那里提交答案

标签： python json pandas apache-spark pyspark

【解决方案1】：

您可以将列合并到地图中，然后从中创建一个 JSON

(df
    .withColumn('json', F.to_json(F.create_map(
        F.lit('name'), F.col('name'),
        F.lit('age'), F.col('age'),
        F.lit('salary'), F.col('salary'),
    )))
    .agg(F.collect_list('json').alias('json_value'))
)

+----------------------------------------------------------------------------------------------------------------------+
|json_value                                                                                                            |
+----------------------------------------------------------------------------------------------------------------------+
|[{"name":"Avery Bradley","age":"25.0","salary":"7730337.0"}, {"name":"Jae Crowder","age":"25.0","salary":"6796117.0"}]|
+----------------------------------------------------------------------------------------------------------------------+

【讨论】：

你能告诉我怎么做吗？根据docs，该列需要是 StructType 或 ArrayType 或 MapType
@pltc - 抱歉，我使用了类似于 mck 答案的结构。 here（colab notebook）就是一个例子。

【解决方案2】：

你可以做一个collect_list的json：

import pyspark.sql.functions as F

df2 = df.agg(F.collect_list(F.to_json(F.struct('*'))).alias('json_data'))

df2.show(truncate=False)
+--------------------------------------------------------------------------------------------------------------+
|json_data                                                                                                     |
+--------------------------------------------------------------------------------------------------------------+
|[{"Name":"Avery Bradley","Age":25.0,"Salary":7730337.0}, {"Name":"Jae Crowder","Age":25.0,"Salary":6796117.0}]|
+--------------------------------------------------------------------------------------------------------------+

【讨论】：