如何处理 AWS Glue 中映射函数中的错误？答案

【问题标题】：How do I handle errors in mapped functions in AWS Glue?如何处理 AWS Glue 中映射函数中的错误？
【发布时间】：2018-11-02 23:29:27
【问题描述】：

我正在使用 DynamicFrame 的 map 方法（或等效的 Map.apply 方法）。我注意到，我传递给这些函数的函数中的任何错误都会被静默忽略，并导致返回的 DynamicFrame 为空。

假设我有一个这样的工作脚本：

import sys
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.transforms import *

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

def my_mapper(rec):
    import logging
    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()

如果我在我的 Glue Dev Endpoint 中使用 gluepython 运行此脚本，我会得到如下输出：

[glue@ip-172-31-83-196 ~]$ gluepython gluejob.py
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/usr/share/aws/glue/etl/jars/glue-assembly.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/usr/lib/spark/jars/slf4j-log4j12-1.7.16.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
18/05/23 20:56:46 WARN Client: Neither spark.yarn.jars nor spark.yarn.archive is set, falling back to uploading libraries under SPARK_HOME.
ERROR StatusLogger No log4j2 configuration file found. Using default configuration: logging only errors to the console.
Count:   0
root

++
||
++
++

关于此输出的说明：

我看不到print 语句或logging.error 语句的结果。
没有迹象表明my_mapper 引发了异常。
printSchema 调用显示生成的 DynamicFrame 上没有架构元数据
show 方法也没有产生任何输出，表明所有行都消失了。

同样，当我将此脚本保存为 AWS Glue 控制台中的作业并运行它时，该作业未指示发生任何错误 - 作业状态为“成功”。值得注意的是，我确实将print 语句和logging.error 调用输出到作业日志，但仅在常规“日志”中，而不是在“错误日志”中。

我想要的是能够表明我的工作失败了，并且能够轻松找到这些错误日志。最重要的是仅仅表明它失败了。

有没有办法在映射函数中记录错误，以便 Glue 将其作为“错误日志”拾取（并将其放在单独的 AWS CloudWatch Logs 路径中）？如果发生这种情况，它会自动将整个 Job 标记为 Failing 吗？还是有其他方法可以从映射函数中显式失败？

（如果有办法记录错误和/或将作业标记为失败，我的计划是创建一个装饰器或其他实用程序函数，它们将自动捕获映射函数中的异常并确保它们被记录和标记失败）。

【问题讨论】：

我不确定如何将 Glue 与 pyspark 一起使用 - 我总是只使用 scala，但您确定您使用的是正确的映射方式吗？ Glue Documentation 描述了使用 Map 类进行转换。
@botchniaque 我已经确认它们是相同的。 Map 只是委托给DynamicFrame.map。我提到它们在我的问题中是等价的。

标签： apache-spark pyspark aws-glue

【解决方案1】：

我发现让 Glue 作业显示为“失败”的唯一方法是从主脚本中引发异常（不是在映射器或过滤器函数中，因为那些似乎得到了旋转到数据处理单元）。

幸运的是，有一种方法可以检测映射或过滤函数内部是否发生异常：使用DynamicFrame.stageErrorsCount() 方法。它将返回一个数字，指示在运行最近的转换时引发了多少异常。

那么解决所有问题的正确方法：

确保您的地图或转换函数明确记录其中发生的任何异常。这最好通过使用装饰器函数或通过其他一些可重用的机制来完成，而不是依赖于在您编写的每个函数中添加 try/except 语句。
在您想要捕获错误的每个转换之后，调用 stageErrorsCount() 方法并检查它是否大于 0。如果您想要中止该作业，只需引发一个异常。

例如：

import logging

def log_errors(inner):
    def wrapper(*args, **kwargs):
        try:
            return inner(*args, **kwargs)
        except Exception as e:
            logging.exception('Error in function: {}'.format(inner))
            raise
    return wrapper

@log_errors
def foo(record):
    1 / 0

然后，在你的工作中，你会做这样的事情：

df = df.map(foo, "foo")
if df.stageErrorsCount() > 0:
    raise Exception("Error in job! See the log!")

请注意，出于某种原因，即使从映射器函数内部调用 logging.exception 仍然不会将日志写入 AWS CloudWatch Logs 中的 error 日志。它被写入常规成功日志。但是，使用这种技术，您至少会看到作业失败并能够在日志中找到信息。另一个警告：Dev Endpoints 似乎没有显示来自映射器或过滤器函数的任何日志。

【讨论】：