PySpark中如何解密Base64加密列及解决自定义函数传参类型不匹配问题

报错原因 #

你编写的decrypt是普通Python函数，只能接收单个字符串类型的输入参数，而col('encrypt')返回的是PySpark的Column表达式对象，代表的是DataFrame的整列数据而非单条值，直接传入当然会触发类型不匹配错误。

解决方案1：注册自定义UDF（通用场景，适配任意解密逻辑） #

你可以把解密函数注册为PySpark可识别的用户自定义函数（UDF），就可以直接作用于整列数据：

import base64
from pyspark.sql.functions import udf, col
from pyspark.sql.types import StringType

# 原有解密逻辑无需修改
def decrypt(trace_info: str):
    msg = base64.urlsafe_b64decode(trace_info)
    # 剩余自定义解密逻辑...
    return trace_info_decrpt

# 注册为UDF，指定返回值为字符串类型
decrypt_udf = udf(decrypt, StringType())

# 调用UDF生成新列
df = df.withColumn('new column name', decrypt_udf(col('encrypt')))

如果你的encrypt列本身不是字符串类型，可以先做类型转换：decrypt_udf(col('encrypt').cast(StringType()))

解决方案2：使用Spark内置函数（性能最优，适配base64解密场景） #

如果你的解密逻辑只有url-safe格式的base64解码，完全可以用Spark原生内置函数实现，规避UDF的跨进程序列化开销，执行效率会高很多：

from pyspark.sql.functions import unbase64, decode, col

# Spark 3.0及以上版本支持urlSafe参数，直接解码后转utf-8字符串
df = df.withColumn('new column name', decode(unbase64(col('encrypt'), urlSafe=True), 'utf-8'))

如果使用的是低于3.0的Spark版本，不支持urlSafe参数，直接使用方案1即可。

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