如何让气流跳过时间表？答案

【问题标题】：How to make airflow skip a schedule?如何让气流跳过时间表？
【发布时间】：2020-01-05 12:06:35
【问题描述】：

我有一个 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

【问题讨论】：

如果当前时间（加上一些调度 delta-T）不在上午 10 点到上午 10:30 之间，您可以使用 PythonOperator 和 skips 开始您的 DAG 跨度>

标签： airflow airflow-scheduler

【解决方案1】：

您可以在工作流开始时使用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,
)

【讨论】：