要完成这个任务，首先需要对输入 Dataframe 进行一些操作，可以按id列进行分组，然后围绕Name_type列进行透视，如下所示：

from pyspark.sql.functions import first

df = spark.createDataFrame(
    [
        ("1", "First", "rob", "Nissan"),
        ("2", "First", "joe", "Hyundai"),
        ("1", "Last", "dent", "Infiniti"),
        ("2", "Last", "Kent", "Genesis")
    ],
    ["id", "Name_type", "Name", "Car"]
)
output = df.groupBy("id").pivot("Name_type").agg(first("Name").alias('Name'), first("Car").alias('Car'))

output.show()
+---+----------+---------+---------+--------+
| id|First_Name|First_Car|Last_Name|Last_Car|
+---+----------+---------+---------+--------+
|  1|       rob|   Nissan|     dent|Infiniti|
|  2|       joe|  Hyundai|     Kent| Genesis|
+---+----------+---------+---------+--------+

然后，您可以使用与您用于获得所需结果的代码完全相同的代码，但使用4列而不是2列：

from pyspark.sql.functions import to_json, struct

column_set = ['First_Name','First_Car', 'Last_Name', 'Last_Car']
output = output.withColumn("json_data", to_json(struct([output[x] for x in column_set])))

output.show(truncate=False)
+---+----------+---------+---------+--------+----------------------------------------------------------------------------------+
|id |First_Name|First_Car|Last_Name|Last_Car|json_data                                                                         |
+---+----------+---------+---------+--------+----------------------------------------------------------------------------------+
|1  |rob       |Nissan   |dent     |Infiniti|{"First_Name":"rob","First_Car":"Nissan","Last_Name":"dent","Last_Car":"Infiniti"}|
|2  |joe       |Hyundai  |Kent     |Genesis |{"First_Name":"joe","First_Car":"Hyundai","Last_Name":"Kent","Last_Car":"Genesis"}|
+---+----------+---------+---------+--------+----------------------------------------------------------------------------------+