如何使用AWS Glue移除S3文件中的非ASCII/不可打印字符？

错误原因 #

你遇到的AttributeError是因为AWS Glue的DynamicFrame不是Spark DataFrame，withColumn是Spark DataFrame专属方法，DynamicFrame不支持该操作。另外你的UDF存在拼写错误：decode('acsii')应改为decode('ascii')，这也会导致后续执行报错。

以下是两种可行的解决方法：

方法一：转换为Spark DataFrame处理 #

先将DynamicFrame转为Spark DataFrame，用Spark API完成清理后再转回DynamicFrame：

1. 修正并定义UDF（增加空值判断避免报错）：

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

def ascii_ignore(x):
    if x is not None:
        # 过滤非ASCII字符，修正拼写错误
        return x.encode('ascii', 'ignore').decode('ascii')
    return x

# 注册UDF并指定返回类型，避免类型推断异常
ascii_udf = udf(ascii_ignore, StringType())

2. 转换并处理数据：

# 将DynamicFrame转为Spark DataFrame
spark_df = mapping_node1694802690218.toDF()

# 清理UPC列的非ASCII/不可打印字符
processed_spark_df = spark_df.withColumn("UPC", ascii_udf(col("UPC")))

# 转回DynamicFrame，适配后续Glue操作流程
processed_dynamic_frame = DynamicFrame.fromDF(processed_spark_df, glueContext, "processed_upc_frame")

方法二：直接处理DynamicFrame（无需转换） #

使用Glue原生的Map.apply方法，直接操作DynamicFrame的每条记录：

1. 定义单条记录的处理函数：

def clean_upc_record(record):
    # 检查字段存在且非空，避免KeyError或None操作报错
    if 'UPC' in record and record['UPC'] is not None:
        record['UPC'] = record['UPC'].encode('ascii', 'ignore').decode('ascii')
    return record

2. 应用处理函数到DynamicFrame：

from awsglue.transforms import Map

processed_dynamic_frame = Map.apply(
    frame=mapping_node1694802690218,
    f=clean_upc_record
)

后续写入S3 #

处理完成后，用Glue API将清理后的文件写入目标S3路径：

glueContext.write_dynamic_frame.from_options(
    frame=processed_dynamic_frame,
    connection_type="s3",
    connection_options={"path": "s3://your-target-bucket/cleaned-data/"},
    format="csv"
)

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