Airflow 循环动态创建的Dataflow任务被标记为Removed状态问题排查

核心矛盾来自对Airflow任务加载机制的误解：Airflow的所有任务、依赖关系必须在Scheduler静态解析DAG文件的阶段完成定义，运行过程中不会动态新增任务。如果后续解析发现元数据库里登记过的任务在当前DAG定义中不存在，就会直接将其标记为Removed状态。
具体到你的代码，存在三处致命错位：

• 动态任务的生成依赖read_text_file()读取/home/airflow/gcs/data/blob_list.txt的结果，但这个文件是get_file_namePythonOperator在DAG真正运行时才会生成的。Scheduler在服务端解析DAG文件时，该路径下根本没有这个文件，全局变量list_of_blobs始终为空，for b in list_of_blobs循环完全不会执行，一个动态Dataflow任务都不会被创建到当前解析的DAG中。
• 你之前本地测试时，本地路径下存在过这个blob列表文件，当时解析生成的所有dataflow_dynamic_*任务被Airflow元数据库留存；部署到线上后，因为线上没有这个静态文件，解析出的DAG仅包含start、get_file_name、end三个任务，元数据库里留存的旧Dataflow任务就全部变成了Removed状态。
• 你目前的逻辑本质是“等DAG运行拿到文件列表再生成任务”，但却把动态任务的创建写在了静态解析阶段的代码块里，完全不符合Airflow的任务加载流程。

根据你的业务场景，有两种可直接落地的实现路径：

路径1：提前可枚举的固定列表场景 #

如果每天需要处理的blob列表可以在Scheduler解析阶段（DAG运行之前）直接从GCS拉取，就把列表拉取逻辑移到DAG定义的全局代码块（with DAG语句之外），在解析阶段就拿到完整的blob列表，再循环生成Dataflow任务。

• 这种方式完全符合Airflow静态解析的要求，所有任务在解析阶段就明确登记到DAG中，不会出现任务缺失的Removed问题
• 注意控制GCS列表拉取的耗时，避免过慢的外部调用拖慢Scheduler的整体解析效率

路径2：运行时才确定的动态列表场景 #

如果blob列表必须等DAG触发、跑完前置PythonOperator才能拿到，直接使用Airflow 2.3+原生支持的**动态任务映射（Dynamic Task Mapping）**能力实现，这也是和你业务目标最匹配的方案：

• 删掉原代码里所有读本地blob文件、写死循环生成Dataflow任务的逻辑
• 改造get_file_name这个PythonOperator的返回值：遍历到的所有符合格式的blob名称，直接打包成列表通过XCom传递，不需要再写入本地文件
• 把原来单个写死参数的DataflowTemplateOperator改成模板化的通用算子，通过.expand()方法绑定上游返回的blob列表：Airflow会在运行时基于上游返回的列表自动生成对应的并行子任务，Scheduler在解析阶段只需要识别到这是一个动态映射任务，不需要提前知道子任务的具体数量和ID，从根本上避免了任务缺失导致的Removed问题。

• 排查时可以先查看Scheduler的解析日志，确认当前解析出的DAG包含哪些task_id，和元数据库里留存的旧task_id做对比，就能快速定位是否是解析阶段未加载到动态任务的问题
• 不要在with DAG的代码块里写依赖DAG运行状态才能拿到的文件、数据的读取逻辑，这部分逻辑必须全部放到运行时的任务执行流程里
• 废弃原来用全局变量传blob列表、写本地临时文件供解析期读取的临时实现，这类写法在多Scheduler副本的部署场景下还会出现文件路径不一致、列表不同步的问题

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