airflow - 如何让气流跳过时间表？

问题描述

我有一个 Airflow DAG，计划在@daily 上午 10 点运行。我想控制我的 DAG 触发器，如果它没有安排在上午 10:00 到上午 10:30 之间，它就不会安排它。

LatestOnlyOperator 无能为力，因为执行持续时间是当天。在我启用调度的那一刻，DAG 将被触发

事实证明，我的流程不是创建一个 30 分钟的窗口，而是需要对手动触发的外部 DAG 的依赖。

让预定 dag - DAG1 ，手动触发 dag - DAG2

这样如果 DAG2 当天被触发。DAG1 应检查执行日期（今天）的最后一个 DAG2 实例状态。

如果成功跳过任务的其余部分

如果 NotExecuted 或 Failed ，运行剩余的任务

我创建了一个分支运算符来执行以下操作。但我无法从DAG2获取task1的最后一个实例。无法正确确定task1的 execution_date 。


branching = BranchPythonOperator(
    task_id='branching',
    python_callable=branchTo,
    dag=dag)


def branchTo(**kwargs):
    date = kwargs['execution_date'] - timedelta(hours=-1)
    dag_folder = conf.get('core', 'DAGS_FOLDER')
    dagbag = DagBag(dag_folder)
    check_dag = dagbag.dags['DAG2']
    my_task = check_dag.get_task('task1')
    ti = TaskInstance(my_task, date)
    state = ti.current_state()
    if state != 'success':
        return 'procees_with_remaining_tasks'
    else:
        return 'mark_success'

由于很难获得手动触发的任务实例。我正在查询一个时间范围以获取为给定的“dag_id”执行的所有 DAG

并使用输出进行条件跳转

@provide_session
def get_task_instances(dag_id, task_id, start_date=None, end_date=None, session=None): 
    TI = TaskInstance
    end_date = end_date or datetime.utcnow()
    list = session.query(TI).filter(
        TI.dag_id == dag_id,
        TI.task_id == task_id,
        TI.execution_date >= start_date,
        TI.execution_date <= end_date,
    ).order_by(TI.execution_date.desc()).all()
    return list

标签： airflowairflow-scheduler

您可以在工作流开始时使用BranchDateTimeOperator来验证当前时间是什么，并根据它决定是继续执行工作流还是分支以结束任务。

from airflow.operators.datetime import BranchDateTimeOperator
cond = BranchDateTimeOperator(
    task_id='datetime_branch',
    follow_task_ids_if_true=['continue_task'],
    follow_task_ids_if_false=['end_task'],
    target_upper=datetime.time(10, 30, 0),
    target_lower=datetime.time(10, 0, 0),
    dag=dag,
)

airflow - 如何让气流跳过时间表？

问题描述

解决方案

推荐阅读