PySpark DataFrame中JSON数组嵌套结构体字段Fernet加密报错求助

错误根因 #

你代码抛出TypeError: encoding without a string argument的核心问题有两个：

1. transform函数迭代数组元素时，你在encrypt中传入了固定的全局列路径字符串"json." + arrayPath + "." + field，该路径指向的是整个数组的字段集合，而非当前迭代到的单个数组元素内的对应字段，导致encrypt接收到的输入不是预期的单个字符串值
2. 若你使用的encrypt是自定义UDF，没有做空值、非字符串类型兼容处理，也会触发Fernet加密时的编码错误

解决步骤 #

1. 先定义兼容异常输入的加密UDF，提前处理空值、非字符串类型输入
2. 在transform的lambda函数中，直接使用迭代变量x引用当前数组元素，通过x[field]获取该元素下的待加密字段值
3. 调整字段更新逻辑，避免全局路径引用错误

修正后代码示例 #

首先注册加密UDF：

from pyspark.sql.functions import udf, lit, transform, col
from cryptography.fernet import Fernet

@udf(returnType="string")
def encrypt(value: str, encrypt_key: str) -> str:
    # 兼容空值、非字符串输入
    if not value or not isinstance(value, str):
        return value
    fernet = Fernet(encrypt_key.encode())
    return fernet.encrypt(value.encode()).decode()

然后处理DataFrame：

full_array_path = f"json.{arrayPath}"
df = df.withColumn(
    "json",
    col("json").withField(
        arrayPath,
        transform(col(full_array_path), lambda x: 
            x.withField(field, encrypt(x[field], lit(encryptKey)))
        )
    )
)

注意事项 #

• 若待加密字段不是字符串类型，需提前做类型转换：encrypt(x[field].cast("string"), lit(encryptKey))
• 若你的json列是JSON字符串类型而非StructType，需要先通过from_json解析为结构体后再做处理，处理完成后可通过to_json转回字符串格式

内容的提问来源于stack exchange，提问作者T.UK