Timetables for complex schedules in Apache Airflow - Mini Project: Build & Apply

Timetables for complex schedules

📖 Scenario: You are managing workflows that need to run at different times during the week. You want to create a schedule that runs tasks on weekdays at 9 AM and on weekends at 10 AM.

🎯 Goal: Build an Airflow DAG timetable that runs at 9 AM on weekdays and 10 AM on weekends using Airflow's Timetable feature.

📋 What You'll Learn

Create a Python dictionary with weekday and weekend times

Define a variable for the timetable class name

Implement a custom timetable class using Airflow's Timetable interface

Print the schedule times generated by the timetable for a sample week

💡 Why This Matters

🌍 Real World

Complex workflows often need different schedules on weekdays and weekends. This project shows how to customize Airflow's scheduling to handle such real-world timing needs.

💼 Career

Understanding how to create custom timetables in Airflow is valuable for DevOps engineers and data engineers who manage workflow automation and scheduling in production environments.

1

Create the schedule times dictionary

Create a dictionary called schedule_times with these exact entries: 'weekday': '09:00' and 'weekend': '10:00'.

Apache Airflow

# Create the schedule_times dictionary with weekday and weekend times
# Your code here

Need a hint?

Use curly braces to create a dictionary with keys 'weekday' and 'weekend' and their respective time strings.

2

Define the timetable class name variable

Define a variable called timetable_class_name and set it to the string 'ComplexScheduleTimetable'.

Apache Airflow

schedule_times = {'weekday': '09:00', 'weekend': '10:00'}
# Define the timetable_class_name variable
# Your code here

Need a hint?

Assign the exact string 'ComplexScheduleTimetable' to the variable timetable_class_name.

3

Implement the custom timetable class

Implement a class called ComplexScheduleTimetable that inherits from airflow.timetables.base.Timetable. Define the method next_dagrun_info that returns the next run time at 9 AM on weekdays and 10 AM on weekends. Use the schedule_times dictionary for the times.

Apache Airflow

schedule_times = {'weekday': '09:00', 'weekend': '10:00'}
timetable_class_name = 'ComplexScheduleTimetable'

from airflow.timetables.base import Timetable
from airflow.utils.dates import cron_presets
from airflow.utils.timezone import datetime, make_aware, timezone
from airflow.utils.dates import DagRunInfo
from datetime import timedelta

# Implement the ComplexScheduleTimetable class here
# Your code here

Need a hint?

Use a loop to find the next date, check if it's weekday or weekend, then set the run time accordingly using schedule_times.

4

Print the next run times for a sample week

Create an instance of ComplexScheduleTimetable called timetable. Then, starting from last_automated_data_interval = null, print the next 7 run start times by calling next_dagrun_info repeatedly and updating last_automated_data_interval with the returned interval. Use print(run_info.data_interval_start) inside the loop.

Apache Airflow

schedule_times = {'weekday': '09:00', 'weekend': '10:00'}
timetable_class_name = 'ComplexScheduleTimetable'

from airflow.timetables.base import Timetable
from airflow.utils.timezone import datetime, make_aware, timezone
from airflow.utils.dates import DagRunInfo
from datetime import timedelta

class ComplexScheduleTimetable(Timetable):
    def next_dagrun_info(self, last_automated_data_interval, restriction):
        from pendulum import datetime as pdatetime
        import pendulum

        if last_automated_data_interval is null:
            next_date = make_aware(datetime.now()).replace(hour=0, minute=0, second=0, microsecond=0)
        else:
            next_date = last_automated_data_interval.end

        # Move to next day
        next_date = next_date + timedelta(days=1)

        # Find next run date and time
        while true:
            weekday = next_date.weekday()  # Monday=0, Sunday=6
            if weekday  make_aware(datetime.now()):
                break
            next_date = next_date + timedelta(days=1)

        return DagRunInfo.interval(run_time, run_time + timedelta(days=1))

# Create timetable instance and print next 7 run start times
# Your code here

Need a hint?

Create an instance of ComplexScheduleTimetable, then loop 7 times calling next_dagrun_info and printing the start time.