如何将PySpark DataFrame转换为嵌套字典：首列为主键，排除空值列

问题还原 #

你已创建如下PySpark DataFrame：

from pyspark.sql.types import StructType, StructField, StringType, IntegerType

data_1 = [
    ("rule1", "", "1", "2", "3", "4"),
    ("rule2", "1", "3", "5", "6", "4"),
    ("rule3", "", "0", "1", "2", "5"),
    ("rule4", "0", "1", "3", "6", "2"),
]

schema = StructType(
    [
        StructField("_c0", StringType(), True),
        StructField("para1", StringType(), True),
        StructField("para2", StringType(), True),
        StructField("para3", StringType(), True),
        StructField("para4", StringType(), True),
        StructField("para5", StringType(), True),
    ]
)
 
df = spark.createDataFrame(data=data_1,schema=schema)

DataFrame内容：

+-----+-----+-----+-----+-----+-----+
|_c0  |para1|para2|para3|para4|para5|
+-----+-----+-----+-----+-----+-----+
|rule1|     |1    |2    |3    |4    |
|rule2|1    |3    |5    |6    |4    |
|rule3|     |0    |1    |2    |5    |
|rule4|0    |1    |3    |6    |2    |
+-----+-----+-----+-----+-----+-----+

需要转换为嵌套字典，要求：

• 外层字典key为_c0的值（如rule1）
• 内层字典仅保留值不为空字符串的列（如rule1不含para1）
• 预期格式：

{
    'rule1': {'para2': '1', 'para3': '2','para4': '3','para5': '4'},
    'rule2': {'para1': '1', 'para2': '3','para3': '5','para4': '6','para5': '4'},
    'rule3': {'para2': '0', 'para3': '1','para4': '2','para5': '5'},
    'rule4': {'para1': '0', 'para2': '1','para3': '3','para4': '6','para5': '2'}
}

错误代码分析 #

你之前的代码：

dict1 = df.rdd.map(lambda row: row.asDict()).collect()
final_dict = {d['_c0']: d[col] for d in dict1 for col in df.columns}

返回结果仅保留每个rule的最后一列值，原因是双重循环的字典推导式会重复覆盖同一个key的value，最终每个rule只保留了最后遍历到的列（para5）的值。

正确解法 #

方法1：使用RDD原生操作（推荐）

通过RDD的map对每一行做过滤处理，再转换为字典：

# 对每一行生成(rule_key, 过滤后的内层字典)的键值对
processed_rdd = df.rdd.map(lambda row: (
    row["_c0"],
    {col: row[col] for col in df.columns if col != "_c0" and row[col] != ""}
))

# 将RDD转换为Python字典
final_dict = dict(processed_rdd.collect())

方法2：借助Pandas处理（适合小数据量）

如果数据量不大，可转成Pandas DataFrame后逐行处理：

import pandas as pd

# 转换为Pandas DataFrame
pd_df = df.toPandas()

final_dict = {}
for _, row in pd_df.iterrows():
    rule_key = row["_c0"]
    # 过滤空值列，生成内层字典
    inner_dict = {col: row[col] for col in pd_df.columns if col != "_c0" and row[col] != ""}
    final_dict[rule_key] = inner_dict

验证结果 #

执行上述代码后，final_dict将完全符合预期格式。

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