在Spark DataFrame中部分替换JSON属性值的实现方案

针对你的需求，推荐使用Spark内置的结构化JSON处理函数结合字符串操作来实现，既满足动态掩码的需求，又能保证性能（避免Python UDF的序列化开销）。以下是具体实现步骤：

步骤1：解析JSON列为结构化类型 #

首先需要将myCol的JSON字符串解析成Spark的StructType，可以通过自动推断或手动定义schema来完成：

from pyspark.sql import functions as F

# 自动推断JSON列的schema（如果结构固定，手动定义schema会更高效）
json_schema = spark.read.json(df.select("myCol").rdd.map(lambda x: x[0])).schema

步骤2：处理指定字段的掩码逻辑 #

对att1和att3字段应用掩码规则：保留前两位字符，剩余部分用-填充（长度与原字符串一致），其他字段保持不变。这里使用substr、concat和lpad组合实现，比regexp_replace更适配动态长度的场景：

processed_df = df.withColumn("parsed_myCol", F.from_json(F.col("myCol"), json_schema)) \
    .withColumn(
        "parsed_myCol",
        F.struct(
            # 处理att1字段
            F.when(
                F.col("parsed_myCol.att1").isNotNull(),
                F.concat(
                    F.substr(F.col("parsed_myCol.att1"), 1, 2),
                    F.lpad("", F.length(F.col("parsed_myCol.att1")) - 2, "-")
                )
            ).otherwise(F.col("parsed_myCol.att1")).alias("att1"),
            # 处理att3字段
            F.when(
                F.col("parsed_myCol.att3").isNotNull(),
                F.concat(
                    F.substr(F.col("parsed_myCol.att3"), 1, 2),
                    F.lpad("", F.length(F.col("parsed_myCol.att3")) - 2, "-")
                )
            ).otherwise(F.col("parsed_myCol.att3")).alias("att3"),
            # 保留其他所有字段不变
            *[F.col(f"parsed_myCol.{col}").alias(col) for col in json_schema.fieldNames() if col not in ["att1", "att3"]]
        )
    ) \
    # 将处理后的结构化列重新序列化为JSON字符串
    .withColumn("myCol", F.to_json(F.col("parsed_myCol"))) \
    .drop("parsed_myCol")

关键说明 #

• 为什么不用regexp_replace？因为regexp_replace的替换字符串无法动态引用原字符串的长度，无法生成对应数量的-，而concat+substr+lpad的组合可以根据原字符串长度动态生成掩码，完全匹配需求。
• 性能优势：全程使用Spark内置函数，避免了Python UDF的跨进程序列化开销，Spark Catalyst优化器可以对这些操作做充分的性能优化，适合大规模数据处理。
• 如果JSON结构存在嵌套，可以通过嵌套struct或array的处理逻辑扩展上述代码，保持同样的性能优势。

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