如何处理AWS Glue映射函数的静默错误并标记任务失败？

我完全懂你的痛点——Glue的map转换默认会悄悄丢弃抛出异常的记录，既不终止任务，也不把错误日志正确上报到CloudWatch的错误路径，排查起来特别闹心。下面是针对这个问题的完整解决方案，包括用装饰器统一处理异常、强制任务失败，以及确保错误日志被正确识别。

问题根源 #

AWS Glue的DynamicFrame.map（以及底层的Map.apply）默认行为是过滤掉所有抛出异常的输入记录，不会触发任务失败。同时，映射函数里的print和普通logging输出运行在Spark Executor节点上，不会同步到Driver节点的控制台日志，所以你在Dev Endpoint或任务控制台看不到这些输出，任务还会显示"成功"。

解决方案：自定义异常处理装饰器 #

我们可以写一个装饰器，自动捕获映射函数中的异常，将错误日志正确输出到CloudWatch错误路径，并抛出Glue的致命异常来终止任务，确保问题被及时发现。

步骤1：实现异常处理装饰器 #

这个装饰器会：

• 捕获映射函数中的所有异常
• 使用Glue官方的日志工具记录错误（确保日志进入CloudWatch错误路径）
• 抛出glue.utils.FatalException，强制任务失败

import logging
from glue.utils import FatalException

def fail_on_exception(func):
    def wrapper(record):
        try:
            return func(record)
        except Exception as e:
            # 获取Glue官方日志器，确保日志进入CloudWatch错误日志路径
            logger = logging.getLogger("glue")
            logger.error(f"[RADIX] Mapper failed for record: {record}. Error: {str(e)}", exc_info=True)
            # 抛出致命异常，强制任务失败
            raise FatalException(f"Mapper failed with error: {str(e)}") from e
    return wrapper

步骤2：修改你的示例脚本 #

将装饰器应用到映射函数上，替换原来的my_mapper：

import sys
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.transforms import *
from glue.utils import FatalException
import logging

glueContext = GlueContext(SparkContext.getOrCreate())
dyF = glueContext.create_dynamic_frame.from_catalog(database="radixdemo", table_name="census_csv")

def fail_on_exception(func):
    def wrapper(record):
        try:
            return func(record)
        except Exception as e:
            logger = logging.getLogger("glue")
            logger.error(f"[RADIX] Mapper failed for record: {record}. Error: {str(e)}", exc_info=True)
            raise FatalException(f"Mapper failed with error: {str(e)}") from e
    return wrapper

@fail_on_exception
def my_mapper(rec):
    logging.error("[RADIX] An error-log from in the mapper!")
    print "[RADIX] from in the mapper!"
    raise Exception("[RADIX] A bug!")

dyF = dyF.map(my_mapper, 'my_mapper')
print "Count: ", dyF.count()
dyF.printSchema()
dyF.toDF().show()

关键说明 #

1. 强制任务失败：使用glue.utils.FatalException而不是普通Exception，因为Glue会识别这个异常并标记任务为失败，而不是继续静默过滤记录。
2. 日志路径：通过logging.getLogger("glue")获取的日志器，会自动将ERROR级别的日志发送到CloudWatch的/aws-glue/jobs/error路径，而不是普通的输出日志。
3. 异常追踪：exc_info=True会将完整的堆栈信息写入日志，方便你定位问题。
4. Dev Endpoint测试：在Glue Dev Endpoint中运行时，你现在会看到明确的异常提示，而不是空的DynamicFrame，同时错误日志会出现在控制台输出里。

额外优化：记录失败的记录 #

如果你希望在终止任务前记录所有失败的记录，可以结合Glue的DynamicFrame来收集错误记录，然后在Driver端判断是否有错误，再决定是否终止任务。比如：

# 新增一个累加器来统计失败记录数
failure_count = sc.accumulator(0)

def fail_on_exception(func):
    def wrapper(record):
        try:
            return func(record)
        except Exception as e:
            global failure_count
            failure_count +=1
            logger = logging.getLogger("glue")
            logger.error(f"[RADIX] Mapper failed for record: {record}. Error: {str(e)}", exc_info=True)
            # 返回None会被过滤，或者你可以返回错误记录到另一个DynamicFrame
            return None
    return wrapper

# 处理数据
dyF_processed = dyF.map(my_mapper, 'my_mapper')

# 在Driver端检查是否有失败记录
if failure_count.value > 0:
    logger = logging.getLogger("glue")
    logger.error(f"[RADIX] Total {failure_count.value} records failed processing")
    raise FatalException(f"Mapper failed for {failure_count.value} records")

这种方式可以先收集所有错误，再一次性终止任务，适合需要统计错误数量的场景。

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