Running tasks in a specific time in airflow

Well, Airflow structure is made so that the schedule_interval is set at the DAG level. This means that you can set the time when an entire DAG will start its execution, but you cannot really specify different execution times per task.

The solution if you have three separate tasks that are not dependent on each other is to create three different DAGs, and schedule them at those three different times.

If instead the time dependency of task_2 and task_3 is not so important, but you only care that are executed one after the other you can, indeed, set dependencies between the tasks so that task_2 runs always after task_1 has finished and task_3 runs always after task_2 has finished. To set dependencies you can use the really handy syntax (assuming your tasks have been assigned to variables task_1, task_2, task_3):

task_1 >> task_2 >> task_3

You can refer to Airflow official documentation for more information.

TL;DR: You cannot schedule single tasks to be run at different specific times, as the only time you can set is the overall DAG run one.

You can use a TimeSensor. Add a time sensor as a sub-task that keeps polling whether the time point has passed. Although it is less elegant, it works.

Airflow Documentation:

https://airflow.apache.org/docs/stable/concepts.html#bitshift-composition

After Airflow 1.8 you can also use the bitshift composition.

Instead of setting a time for your tasks, use this approach:

op1 = DummyTask(...)
op2 = DummyTask(...)

op1 >> op2 # same as: op1.set_downstream(op2)

The assignment above means that Airflow will only execute op2 after op1 is successfully completed.

We usually write a help function like:

def wait_till(hour: int, minute: int, second: int, dag):
    """get a DateTimeSensor runs till hour: minute: second for default timezone

    Parameters
    ----------
    hour : int
        hour in the day
    minute : int
        minute
    second : int
        second
    dag : [type]
        dag
    """
    target_time_str = f'next_execution_date.in_tz("Asia/Shanghai").replace(hour={hour}, minute={minute}, second={second})'
    task_id_str = f"wait_till_{hour:02d}{minute:02d}{second:02d}"
    return DateTimeSensor(
        task_id=task_id_str, target_time="{{ " + target_time_str + " }}", dag=dag, poke_interval=5
    )

Then use this as a timer task inside a dag like

wait_till(8,0,0) >> your_task

We also tried putting the single task in a separate dag and then use external task sensor to model the task dependencies. The problem with this approach for us is that we often need to reschedule the task. Rescheduling a dag in airflow would lose all of the logs, which is not acceptable for us.