Airflow TriggerDagRunOperator并行运行时随机失败问题求助

问题背景 #

使用TriggerDagRunOperator实现跨DAG触发工作流，串行链式执行（trigger_1 >> trigger_2 >> ...）完全正常，但切换为并行调度模式（[trigger_1, trigger_2] >> trigger_3 >> [trigger_4, trigger_5]）后，触发任务会随机失败——即便子DAG最终以all_success状态完成。

触发任务无有效业务日志，仅返回系统提示：

*** The task might not have been executed, logs were deleted as part of logs retention (default of 30 days), or worker executing it might have finished abnormally (e.g. was evicted).

已确认基础环境无资源瓶颈：

• 运行时CPU、内存使用率均低于50%
• Airflow配置max_active_runs_per_dag=25、dag_concurrency=100，仅并行触发2个子DAG
• 后续定位到核心诱因：子DAG中存在任务重试记录时，TriggerDagRunOperator会直接标记自身失败

解决方法 #

1. 修复子DAG重试导致的触发任务失败 #

默认的TriggerDagRunOperator在wait_for_completion=True时，会检查子DAG所有任务实例的状态历史，只要有任务触发过重试，就会判定子DAG运行异常。可以通过自定义Operator规避这个逻辑：

from airflow.operators.trigger_dagrun import TriggerDagRunOperator
from airflow.utils.state import State

class CustomTriggerDagRunOperator(TriggerDagRunOperator):
    def _check_trigger_status(self, dag_run):
        # 仅校验子DAG的最终运行状态，忽略任务重试记录
        if dag_run.state == State.SUCCESS:
            return True
        elif dag_run.state in (State.FAILED, State.UP_FOR_RETRY, State.UP_FOR_RESCHEDULE):
            return False
        # 返回None表示继续轮询状态
        return None

将原代码中的TriggerDagRunOperator替换为这个自定义类，即可避免子DAG任务重试导致的触发任务失败。

如果业务场景允许，也可以直接将子DAG中任务的retries参数设为0，从根源消除重试记录，但此方案局限性较强，仅适用于无需重试的业务。

2. 解决并行调度下的随机失败 #

除了子DAG重试的因素，并行触发时的随机失败还可以从以下方向排查：

• Worker进程稳定性：查看Airflow Worker的系统日志（如/var/log/airflow/worker.log），确认是否存在进程被OOM Killer终止、节点网络波动导致的Worker失联情况。
• 状态轮询间隔优化：当前设置poke_interval=240（4分钟），如果子DAG完成速度较快，可能存在元数据库状态同步延迟，导致Operator误判。可将poke_interval缩小至60秒，提升状态检查频率。
• 元数据库锁竞争：并行触发多个DAG时，元数据库可能出现锁等待，导致状态查询超时。检查数据库慢查询日志，必要时调整数据库连接数或优化表索引。

3. 日志缺失问题处理 #

如果触发任务无日志输出，可按以下步骤排查：

• 检查Airflow配置log_retention_days，确认是否日志被提前清理（默认30天，测试场景可临时调大）。
• 确认Worker节点的日志存储目录权限，确保Worker进程有写入日志的权限。

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