PySpark如何不使用UDF实现DataFrame非空列拼接为JSON列

需求为对任意结构的Spark DataFrame新增JSON列，核心规则：

• JSON的键为原始DataFrame的列名，值为对应行的列取值
• 自动过滤值为空的字段，不写入JSON
• 列名、列总数动态变化，不支持硬编码枚举所有列

输入样例 #

输出预期 #

新增JSON列，仅保留非空字段键值对：

当前基于Python UDF的实现可满足功能，但性能损耗明显，代码如下：

from pyspark.sql.functions import udf, struct
from pyspark.sql.types import StringType
import json

def jsonize_fields(row):
    field_dict = {}
    for k, v in row.asDict().items():
        if v:
            field_dict[k] = v
    return json.dumps(field_dict)

jsonize_udf = udf(jsonize_fields, StringType())
spark_data_frame = spark_data_frame.withColumn('JSON',
jsonize_udf(struct(*spark_data_frame.columns)))

需要实现无UDF依赖的原生Spark高性能方案。

直接使用Spark内置函数实现，无Python进程序列化/反序列化开销，性能远高于UDF实现：

from pyspark.sql.functions import to_json, when, col, lit, create_map

# 动态拼接所有列的键值对表达式，空值位置传null供后续过滤
kv_exprs = []
for col_name in spark_data_frame.columns:
    kv_exprs.append(lit(col_name))
    # 可根据业务空值规则调整判断条件，比如同时过滤空字符串可加 `& (col(col_name) != "")`
    kv_exprs.append(when(col(col_name).isNotNull(), col(col_name)))

result_df = spark_data_frame.withColumn(
    "JSON",
    to_json(create_map(*kv_exprs))
)

实现原理：Spark内置to_json方法处理Map类型数据时，会自动丢弃值为null的键值对，天然匹配过滤空字段的需求，全程由Spark Catalyst优化器生成执行计划，无额外性能损耗。

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