能否利用Airflow标签判断指定标签的所有DAG是否完成并触发新任务？

完全可以通过US_TAG标签实现这个需求，核心是利用Airflow的状态监听和标签筛选能力，下面给你两种实用的实现思路：

方法一：用自定义传感器监听所有带标签DAG的运行状态 #

你可以写一个PythonSensor，通过查询Airflow的元数据，筛选出所有带有US_TAG标签的DAG，然后检查它们的最新运行实例是否都处于成功状态。一旦全部成功，传感器就会触发后续任务。

示例代码片段：

from airflow.sensors.python import PythonSensor
from airflow.models import DagTag, DagRun
from airflow.utils.state import State

def check_all_tagged_dags_success(**context):
    # 获取所有带有US_TAG的DAG ID
    tagged_dag_ids = [dag_tag.dag_id for dag_tag in DagTag.filter(DagTag.tag == 'US_TAG')]
    if not tagged_dag_ids:
        return False
    # 检查每个DAG的最新运行实例状态
    for dag_id in tagged_dag_ids:
        latest_run = DagRun.find(dag_id=dag_id, state=State.SUCCESS, order_by=DagRun.execution_date.desc(), limit=1)
        if not latest_run:
            # 该DAG没有成功的运行实例，或者还没跑完
            return False
    return True

# 定义传感器任务
wait_for_all_us_dags = PythonSensor(
    task_id='wait_for_all_us_dags',
    python_callable=check_all_tagged_dags_success,
    poke_interval=60,  # 每分钟检查一次
    mode='reschedule',
    dag=your_target_dag  # 包含后续任务的DAG
)

# 后续任务依赖这个传感器
your_post_task.set_upstream(wait_for_all_us_dags)

方法二：用回调+全局计数器触发任务 #

给每个带有US_TAG的DAG添加成功回调，每次DAG运行成功后，就更新一个全局计数器（比如用Airflow的Variable或者外部缓存如Redis）。当计数器的值等于带标签的DAG总数时，触发后续任务。

步骤：

1. 先统计带US_TAG的DAG总数，存入Airflow Variable（比如us_dag_total_count）
2. 给每个带标签的DAG添加成功回调：

from airflow.models import Variable

def update_success_counter(**context):
    current_count = Variable.get('us_dag_success_count', default_var=0)
    Variable.set('us_dag_success_count', int(current_count) + 1)
    # 检查是否达到总数
    total_count = int(Variable.get('us_dag_total_count'))
    if int(current_count) + 1 == total_count:
        # 触发后续任务，比如通过Airflow API或者直接调用任务逻辑
        trigger_your_post_task()
        # 重置计数器，避免下一次运行重复触发
        Variable.set('us_dag_success_count', 0)

# 在每个带US_TAG的DAG中添加
dag = DAG(
    dag_id='your_us_dag',
    tags=['US_TAG'],
    on_success_callback=update_success_counter,
    # 其他DAG配置
)

注意事项 #

• 要确保所有带US_TAG的DAG的运行周期是对齐的，避免跨周期的运行实例干扰判断
• 如果DAG数量会动态变化，方法一中的tagged_dag_ids会自动获取最新列表，方法二则需要定期更新us_dag_total_count变量
• 方法一中的数据库查询要注意性能，避免高频率查询导致元数据库压力过大
• 两种方法都要处理DAG失败的情况：比如方法一要考虑如果某个DAG运行失败，传感器应该持续等待或者触发告警；方法二要在DAG失败时重置计数器，避免计数卡住

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