Airflow 2.7.2中EmrContainerOperator任务失败程序化重触发方案

方案一：利用Airflow原生重试参数（最简单） #

Airflow任务本身支持重试配置，你可以通过设置retries为极大值模拟无限重试，并配置retry_delay控制重试间隔，直接实现失败即自动重试的逻辑：

from airflow.providers.amazon.aws.operators.emr_containers import EmrContainerOperator
from datetime import timedelta

emr_spark_task = EmrContainerOperator(
    task_id="emr_structured_streaming_job",
    job_name="structured_streaming_job",
    virtual_cluster_id="your_virtual_cluster_id",
    execution_role_arn="your_execution_role_arn",
    release_label="emr-6.10.0",
    job_driver={
        "sparkSubmitJobDriver": {
            "entryPoint": "s3://your-bucket/path/to/streaming-job.py",
            "entryPointArguments": ["--param1", "value1"]
        }
    },
    configuration_overrides={
        "applicationConfiguration": [
            {
                "classification": "spark-defaults",
                "properties": {
                    "spark.sql.streaming.checkpointLocation": "s3://your-bucket/checkpoint"
                }
            }
        ]
    },
    # 核心重试配置
    retries=99999,  # 设为极大值模拟无限重试
    retry_delay=timedelta(seconds=30),  # 每次重试间隔30秒，可按需调整
    retry_exponential_backoff=False,  # 关闭指数退避，保持固定间隔
)

这种方式依赖Airflow原生能力，无需额外开发，适合大多数场景。

方案二：自定义失败回调实现精准重触发（更灵活） #

如果需要在重试前执行额外逻辑（比如日志上报、资源检查），可以通过on_failure_callback自定义函数，调用Airflow内部API重新触发当前任务实例：

from airflow.providers.amazon.aws.operators.emr_containers import EmrContainerOperator
from airflow.models import TaskInstance
from airflow.utils.session import create_session
from datetime import timedelta
import logging

def retrigger_task(context):
    ti = context["task_instance"]
    logging.info(f"任务 {ti.task_id} 失败，准备重新触发")
    
    with create_session() as session:
        # 创建新的任务实例并设为待运行状态
        new_ti = TaskInstance(
            task=ti.task,
            execution_date=ti.execution_date,
            state="queued"
        )
        session.merge(new_ti)
        session.commit()
    logging.info(f"已成功重新触发任务 {ti.task_id}")

emr_spark_task = EmrContainerOperator(
    task_id="emr_structured_streaming_job",
    # 其他EmrContainerOperator参数省略...
    on_failure_callback=retrigger_task,
    retries=0  # 关闭原生重试，避免与自定义回调冲突
)

这个方案的优势是可以在重试前后加入自定义逻辑，比如检查EMR集群状态、清理临时文件等，完全由你控制触发时机。

注意事项 #

• 无限重试可能导致资源耗尽，建议添加监控告警（比如任务连续失败N次后发送通知），避免无意义的循环。
• 结构化Spark Streaming作业必须配置checkpointLocation，确保重试时能从断点继续处理，避免数据重复或丢失。
• Airflow 2.7.2中，EmrContainerOperator的重试逻辑会基于EMR作业的最终失败状态触发，无需额外判断作业状态。

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