迭代数据框的所有行并从其数据中创建新对象答案

【问题标题】：Iterate all the rows of data frame and create new objects from its data迭代数据框的所有行并从其数据中创建新对象
【发布时间】：2021-04-08 21:43:49
【问题描述】：

我有一个包含多列的数据框，我想为数据框中的每一行创建一个 JSON，其中包含特定列的值。

说，我有一个数据框df1，其中列有A、B、C、D、E、F、G。

我正在尝试创建 JSON 结构

{
   "source_id": df1.C,
   "target_id": df1.G,
   "time_of_creation": 123456789123
}

我目前正在使用 df1.rdd.toLocalIterator 遍历每一行

如果有更好的方法请告诉我

【问题讨论】：

所以你想创建list of json（每行一个）？或者您想创建 pyspark 数据框，具有 json 格式的单列（一行一 json），或者您想将整个数据框转换为单个 json 文件？
是的，没错。我想创建一个 JSON 列表（每行一个）。因为这个列表被用作另一个休息请求 (POST) 的输入。

标签： json apache-spark pyspark apache-spark-sql

【解决方案1】：

您可以使用to_json 和named_struct：

df1.show()
+---+---+
|  C|  G|
+---+---+
| c1| g1|
+---+---+

result = df1.selectExpr("""
    to_json(
        named_struct(
            'source_id', C,
            'target_id', G,
            'time_of_creation', current_timestamp()
        )
    ) output_json
""")

result.show(20,0)
+---------------------------------------------------------------------------------+
|output_json                                                                      |
+---------------------------------------------------------------------------------+
|{"source_id":"c1","target_id":"g1","time_of_creation":"2021-01-02T14:12:19.612Z"}|
+---------------------------------------------------------------------------------+

【讨论】：

是否可以将此输出到另一个字符串变量？因为我基本上想创建一个 JSON 列表（每行一个），因为该列表将用作另一个休息请求 (POST) 的输入。
@Divakar 你可以使用jsonlist = result.rdd.map(lambda r: r[0]).collect()
我被要求不要使用“collect()”或“toLocalIterator()”，因为它不是并行计算中火花基础设施的最佳使用。请纠正我，如果我对“收集”的理解不正确，会将所有记录带入驱动程序，因此我们正在有效地按顺序处理它们。
@Divakar 你说你想收集一个列表，所以你需要做 collect 以便将所有已处理的记录带回给驱动程序。其他操作并行完成，但最后一步必须涉及收集，以便将您想要的列表返回到驱动程序中。