如何将PySpark DataFrame的行转换为指定字典格式输出？

1. 先对齐目标数据结构 #

先完成你提到的类型转换和列名调整（比如将timestamp转为Timestamp类型、total_price转为Double类型，把User重命名为user，同时确保type字段值仅保留ADDED_TO_CART或APP_OPENED），得到结构匹配目标要求的enrich_clean DataFrame。

2. 转换为目标字典格式的两种方法 #

方法1：RDD映射+row.asDict() #

你用到的enrich.rdd.map(lambda row: row.asDict())是直接有效的方式，它会把PySpark的Row对象转为标准Python字典：

dict_rdd = enrich_clean.rdd.map(lambda row: row.asDict())

方法2：JSON转字典（适合需要序列化场景） #

如果需要先转成JSON格式再解析为字典，可以用toJSON()配合json.loads：

import json
dict_list = enrich_clean.toJSON().map(lambda json_str: json.loads(json_str)).collect()

3. 查看转换后的结果 #

PySpark的RDD是懒执行的，仅定义转换操作不会触发计算，必须调用动作算子才能看到结果：

• 获取全部结果（数据量大时谨慎使用）：

full_result = dict_rdd.collect()
for item in full_result:
    print(item)

• 获取前N条样本（适合大数据集）：

sample_result = dict_rdd.take(5)
print(sample_result)

• 逐条打印结果：

dict_rdd.foreach(print)

完整示例代码 #

# 补充类型转换和列名调整的示例代码
from pyspark.sql.functions import to_timestamp, col
from pyspark.sql.types import DoubleType

# 处理数据类型与列名
enrich_clean = enrich \
    .withColumn("timestamp", to_timestamp(col("timestamp"))) \
    .withColumn("total_price", col("total_price").cast(DoubleType())) \
    .withColumnRenamed("User", "user") \
    .filter(col("type").isin("ADDED_TO_CART", "APP_OPENED"))

# 转换为字典并查看前5条样本
sample_dicts = enrich_clean.rdd.map(lambda row: row.asDict()).take(5)
for d in sample_dicts:
    print(d)

内容的提问来源于stack exchange，提问作者RustyShackleford