AWS MWAA调用EMR Serverless长时同步任务遇Airflow重试上限错误如何解决？

错误原因 #

该错误源于Airflow的AWS等待器（waiter_with_logging.py）在等待EMR Serverless任务完成时，耗尽了预设的最大重试次数。默认情况下，Airflow对EMR Serverless任务的等待配置（重试间隔、最大尝试次数）无法覆盖长时运行任务的耗时需求。

解决方法 #

1. 调整EMR Serverless Operator的等待参数 #

使用EmrServerlessStartJobOperator时，显式配置waiter_delay（重试间隔，单位秒）和waiter_max_attempts（最大重试次数），确保两者乘积（总超时时间）大于任务的最长运行时长。

示例代码：

from airflow.providers.amazon.aws.operators.emr_serverless import EmrServerlessStartJobOperator

run_emr_serverless_job = EmrServerlessStartJobOperator(
    task_id="run_emr_serverless_job",
    application_id="your-emr-serverless-app-id",
    execution_role_arn="your-execution-role-arn",
    job_driver={
        "sparkSubmit": {
            "entryPoint": "s3://your-bucket/your-job.py",
            "entryPointArguments": ["arg1", "arg2"]
        }
    },
    # 示例：间隔300秒（5分钟）重试一次，最大72次，总超时360分钟（6小时）
    waiter_delay=300,
    waiter_max_attempts=72,
)

2. 拆分长任务（可选） #

若任务运行时长超过8小时，可将其拆分为多个短任务，通过Airflow的任务依赖串联执行，规避单个等待器的超时限制。

3. 检查MWAA环境超时配置 #

确认MWAA环境的全局任务超时参数（如core.dag_run_timeout）是否足够，避免DAG整体超时覆盖EMR任务的等待窗口，可在MWAA环境配置页面调整相关参数。

4. 切换为异步等待模式（替代方案） #

若同步等待非必需，可改为提交任务后立即返回，后续通过EmrServerlessJobSensor轮询任务状态，灵活控制检查间隔和超时时间。

示例代码：

from airflow.providers.amazon.aws.operators.emr_serverless import EmrServerlessStartJobOperator
from airflow.providers.amazon.aws.sensors.emr_serverless import EmrServerlessJobSensor

start_job = EmrServerlessStartJobOperator(
    task_id="start_emr_serverless_job",
    application_id="your-app-id",
    execution_role_arn="your-role-arn",
    job_driver={...},
    wait_for_completion=False,  # 提交任务后不等待完成
)

wait_for_job = EmrServerlessJobSensor(
    task_id="wait_for_emr_serverless_job",
    application_id="your-app-id",
    job_run_id="{{ task_instance.xcom_pull(task_ids='start_emr_serverless_job') }}",
    poke_interval=300,  # 每5分钟检查一次任务状态
    timeout=21600,  # 总超时6小时
)

start_job >> wait_for_job

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