PySpark如何对DataFrame筛选后指定列应用自定义解密函数并导出到S3

推荐通过PySpark内置的用户自定义函数（UDF）实现对列的批量自定义转换，这是符合Spark分布式计算逻辑的标准实现方式，无需转为RDD也不需要把数据拉到Driver端。

完整实现步骤 #

1. 导入依赖包 #

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

2. 将自定义解密函数注册为Spark UDF #

需要指定返回值类型，你的解密函数返回字符串，因此指定为StringType()：

decrypt_udf = udf(decrypt, StringType())

3. 应用UDF处理目标数据 #

过滤version=1的行后，直接对body列应用UDF得到解密结果：

# 得到存储了解密后内容的DataFrame
decrypted_df = df.where(col('version') == '1') \
                 .select(decrypt_udf(col('body')).alias('decrypted_body'))

# 可执行show查看结果验证逻辑
decrypted_df.show()

4. 写入S3为CSV格式 #

直接调用DataFrame的write接口即可分布式写入目标路径：

decrypted_df.write \
            .option("header", True)  # 若不需要表头可改为False
            .mode("overwrite")       # 写入模式：overwrite覆盖已有数据、append追加、ignore存在则跳过
            .csv("s3://<你的S3存储桶名称>/<自定义输出目录路径>")

补充说明：之前RDD写法报错的原因 #

你之前调用.rdd.map失败的原因是RDD中的每个元素是Row对象，不是直接的body列字符串值，如果一定要用RDD实现需要先取出对应字段值，写法如下：

decrypted_df = df.where(col('version') == '1') \
                 .select(col('body')) \
                 .rdd.map(lambda row: decrypt(row['body'])) \
                 .toDF(['decrypted_body'])

该方式性能低于直接使用DataFrame UDF，更推荐优先使用UDF方案。

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