Apache Airflow自触发DAG 8:00-21:00调度实现方案求助

你可以通过两种常用的方案实现自触发DAG的时段限制，不需要修改现有的自触发核心逻辑：

方案1：新增ShortCircuitOperator做时段校验 #

在触发自身的任务前新增一个短路任务，判断当前执行时间是否在每日8:00-21:00的范围内，只有符合条件才会继续执行触发下一次DAG运行的任务，不符合就终止当前链路，不会触发下一次运行，自然就会停止到次日早8点后再启动。
修改后的代码示例：

from datetime import timedelta, datetime
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.dagrun_operator import TriggerDagRunOperator
from airflow.operators.python_operator import ShortCircuitOperator

# 时间校验函数，判断当前时间是否在8-21点之间
def check_run_time(**context):
    current_hour = context['execution_date'].hour
    return 8 <= current_hour <= 21


default_args = {
    'owner': 'ownername',
    'depends_on_past': False,
    'start_date': datetime(2021,3,2,10,1),
    'email': ['***@mail.com'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}
with DAG('self_trigger_dag', schedule_interval=None, max_active_runs=1, catchup=False, default_args=default_args) as dag:
    sleep_task = BashOperator(
        task_id='sleep_task',
        bash_command='sleep 180',
        dag=dag,
    )

    run_command = BashOperator(
        task_id='run_command',
        bash_command="my bash_command",
        use_legacy_sql=False,
        dag=dag,
    )

    # 新增时间校验任务
    check_time_task = ShortCircuitOperator(
        task_id='check_run_time',
        python_callable=check_run_time,
        provide_context=True,
        dag=dag
    )
    
    dag_trigger = TriggerDagRunOperator(
        task_id='trigger_self',
        trigger_dag_id='self_trigger_dag',
        dag=dag)

    sleep_task >> run_command >> check_time_task >> dag_trigger

这个方案的优势是逻辑简单，对原有DAG改动很小，不会产生多余的DAG运行记录。

方案2：给TriggerDagRunOperator指定偏移后的运行时间 #

如果你希望到次日早8点自动触发下一次运行，不需要手动启动首次运行，可在触发任务中指定下一次的运行时间为符合时段要求的时间点，当当前时间超过21点时，把下一次触发时间设置为次日8点：

from datetime import timedelta, datetime
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.dagrun_operator import TriggerDagRunOperator

default_args = {
    'owner': 'ownername',
    'depends_on_past': False,
    'start_date': datetime(2021,3,2,10,1),
    'email': ['***@mail.com'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

def get_next_execution_date(current_dt):
    if current_dt.hour < 8:
        # 早于8点，设置为当天8点运行
        return current_dt.replace(hour=8, minute=0, second=0, microsecond=0)
    elif current_dt.hour > 21:
        # 晚于21点，设置为次日8点运行
        return (current_dt + timedelta(days=1)).replace(hour=8, minute=0, second=0, microsecond=0)
    else:
        # 时段内，立即触发下一次
        return current_dt

with DAG('self_trigger_dag', schedule_interval=None, max_active_runs=1, catchup=False, default_args=default_args) as dag:
    sleep_task = BashOperator(
        task_id='sleep_task',
        bash_command='sleep 180',
        dag=dag,
    )

    run_command = BashOperator(
        task_id='run_command',
        bash_command="my bash_command",
        use_legacy_sql=False,
        dag=dag,
    )
    
    dag_trigger = TriggerDagRunOperator(
        task_id='trigger_self',
        trigger_dag_id='self_trigger_dag',
        execution_date="{{ get_next_execution_date(execution_date) }}",
        dag=dag)

    sleep_task >> run_command >> dag_trigger

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