Apache托管Airflow中使用EMR Operator的DAG执行失败问题咨询

结论 #

该报错和Airflow本身无关，是MWAA托管环境配置、任务参数设置不合理导致的。

报错核心特征 #

你提供的报错日志如下：

[2021-08-25 05:00:04,520] {{logging_mixin.py:104}} INFO - [2021-08-25 05:00:04,520] {{local_task_job.py:188}} WARNING - State of this instance has been externally set to skipped. Terminating instance.
[2021-08-25 05:00:04,600] {{process_utils.py:100}} INFO - Sending Signals.SIGTERM to GPID 1897
[2021-08-25 05:00:04,621] {{taskinstance.py:1265}} ERROR - Received SIGTERM. Terminating subprocesses.
[2021-08-25 05:00:04,702] {{taskinstance.py:1482}} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1138, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1311, in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context, task_copy)
  File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1341, in _execute_task
    result = task_copy.execute(context=context)
  File "/usr/local/lib/python3.7/site-packages/airflow/sensors/base.py", line 243, in execute
    time.sleep(self._get_next_poke_interval(started_at, run_duration, try_number))
  File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1267, in signal_handler
    raise AirflowException("Task received SIGTERM signal")
airflow.exceptions.AirflowException: Task received SIGTERM signal
[2021-08-25 05:00:04,725] {{taskinstance.py:1532}} INFO - Marking task as FAILED. dag_id=emr_job_flow_manual_steps_dag, task_id=watch_step, execution_date=20210824T030000, start_date=20210825T030008, end_date=2021-08-25 05:00:04
[2021-08-25 05:00:04,793] {{process_utils.py:66}} INFO - Process psutil.Process(pid=1897, status='terminated', exitcode=1, started='03:00:07') (1897) terminated with exit code 1

核心触发信号是MWAA管控平面主动发送的SIGTERM，明确提示任务状态被外部设置为skipped后终止。

问题根因 #

• 超时设置不合理：报错的watch_step是EMR步骤监听Sensor任务，从日志可见该任务运行了近2小时后被终止，要么是DAG的dagrun_timeout参数设置过短，要么是Sensor本身的timeout参数小于EMR作业实际运行时长，触发了超时强制终止逻辑。
• Sensor运行模式不匹配：默认的poke模式Sensor会持续占用Worker进程直到监听完成，长时间占用会导致MWAA的资源管控逻辑主动终止进程，释放资源。
• mw1.small规格限制：该规格是MWAA最小入门规格，Worker资源配额低、任务最大允许运行时长短，对于运行时间超过1小时的长任务，很容易触发资源回收机制。

解决方案 #

• 调整超时参数：根据你的EMR作业最长运行时间，给DAG设置足够长的dagrun_timeout，同时给watch_step的EmrStepSensor设置对应的timeout参数，避免超时被终止。
• 更换Sensor运行模式：给EmrStepSensor添加mode="reschedule"参数，每次检查完步骤状态后释放Worker资源，到下一次检查时间再重新调度，不会长期占用进程。
• 升级MWAA规格：如果作业普遍运行时长超过2小时，建议将环境规格升级为mw1.medium，获得更高的资源配额和更长的任务运行允许时长。
• 排查外部操作：确认没有手动修改任务状态、手动停止DAG运行的操作，也没有其他依赖任务异常触发了跳过当前任务的规则。

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