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

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

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

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

Hint

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

Practice

1. What is the main purpose of a timetable in Apache Airflow?

easy

A. To control when workflows (DAGs) are scheduled to run

B. To store data processed by workflows

C. To define the tasks inside a workflow

D. To monitor the status of running tasks

Solution

1. Step 1: Understand Airflow scheduling

   Airflow uses timetables to decide when to start running workflows (DAGs).

2. Step 2: Differentiate timetable from other components

   Timetables do not store data, define tasks, or monitor tasks; they only schedule runs.

3. Final Answer:

   To control when workflows (DAGs) are scheduled to run -> Option A

4. Quick Check:

   Timetable = schedule control [OK]

Hint: Timetables decide when workflows start running [OK]

Common Mistakes:

• Confusing timetable with task definitions
• Thinking timetable stores data
• Mixing timetable with monitoring features

2. Which of the following is the correct way to import the built-in cron timetable in Airflow 2.5+?

easy

A. import airflow.timetables.Cron as CronTimetable

B. import airflow.timetable.cron as CronTimetable

C. from airflow.schedules import CronTimetable

D. from airflow.timetables.cron import CronTimetable

Solution

1. Step 1: Recall Airflow timetable import paths

   Airflow 2.5+ uses airflow.timetables.cron for cron timetable classes.

2. Step 2: Check syntax correctness

   from airflow.timetables.cron import CronTimetable uses correct module and class name with proper Python import syntax.

3. Final Answer:

   from airflow.timetables.cron import CronTimetable -> Option D

4. Quick Check:

   Correct import path = from airflow.timetables.cron import CronTimetable [OK]

Hint: Use airflow.timetables.cron for cron timetable imports [OK]

Common Mistakes:

• Using wrong module name like 'schedules' instead of 'timetables'
• Incorrect capitalization in module or class names
• Wrong import syntax causing errors

3. Given this Airflow timetable code snippet, what schedule does it represent?

from airflow.timetables.cron import CronTimetable
cron = CronTimetable('0 6 * * 1-5')

medium

A. Runs every day at 6:00 AM

B. Runs every hour from 6 AM to 5 PM

C. Runs at 6:00 AM every weekday (Monday to Friday)

D. Runs only on weekends at 6:00 AM

Solution

1. Step 1: Interpret the cron expression '0 6 * * 1-5'

   The fields mean: minute=0, hour=6, day=any, month=any, weekday=1-5 (Monday to Friday).

2. Step 2: Match cron to schedule meaning

   This means the job runs at 6:00 AM on weekdays only.

3. Final Answer:

   Runs at 6:00 AM every weekday (Monday to Friday) -> Option C

4. Quick Check:

   Cron '0 6 * * 1-5' = Weekday 6 AM [OK]

Hint: Cron weekdays 1-5 means Monday to Friday [OK]

Common Mistakes:

• Assuming it runs every day instead of weekdays only
• Confusing hour and minute fields
• Thinking it runs on weekends

4. You wrote this custom timetable code but Airflow raises an error:

class MyTimetable:
    def __init__(self, cron):
        self.cron = cron

my_tt = MyTimetable('0 0 * * *')

What is the likely cause of the error?

medium

A. Not importing CronTimetable from airflow

B. Missing required methods like 'next_dagrun_info' in the timetable class

C. Using a class instead of a function for timetable

D. Incorrect cron string format

Solution

1. Step 1: Understand Airflow timetable requirements

   Custom timetables must implement specific methods like 'next_dagrun_info' to work.

2. Step 2: Check provided class

   The class only stores cron string but lacks required methods, causing Airflow error.

3. Final Answer:

   Missing required methods like 'next_dagrun_info' in the timetable class -> Option B

4. Quick Check:

   Custom timetable needs required methods [OK]

Hint: Custom timetables must implement Airflow's required methods [OK]

Common Mistakes:

• Assuming any class with cron string works as timetable
• Blaming cron string format without checking methods
• Confusing timetable class with task definitions

5. You want to run a DAG every Monday at 8 AM and also on the 1st day of every month at 9 AM. Which Airflow timetable setup correctly combines these schedules?

hard

A. Use a custom timetable that combines two CronTimetable instances for '0 8 * * 1' and '0 9 1 * *', then merges their schedules

B. Create two DAGs, one with '0 8 * * 1' and another with '0 9 1 * *', then combine their runs manually

C. Use a single CronTimetable with expression '0 8 * * 1,0 9 1 * *'

D. Set schedule_interval='@weekly' and add a sensor to trigger on the 1st day of month

Solution

1. Step 1: Understand the scheduling needs

   We need two different schedules: Mondays at 8 AM and 1st day of month at 9 AM.

2. Step 2: Evaluate options for combining schedules

   A single cron expression cannot represent both times simultaneously. Two DAGs is possible but not ideal. A custom timetable combining two CronTimetable instances is the correct approach.

3. Step 3: Why other options fail

   Use a single CronTimetable with expression '0 8 * * 1,0 9 1 * *' is invalid cron syntax. Create two DAGs, one with '0 8 * * 1' and another with '0 9 1 * *', then combine their runs manually is manual and complex. Set schedule_interval='@weekly' and add a sensor to trigger on the 1st day of month uses sensors which is indirect and less clean.

4. Final Answer:

   Use a custom timetable that combines two CronTimetable instances for '0 8 * * 1' and '0 9 1 * *', then merges their schedules -> Option A

5. Quick Check:

   Combine schedules with custom timetable = Use a custom timetable that combines two CronTimetable instances for '0 8 * * 1' and '0 9 1 * *', then merges their schedules [OK]

Hint: Combine multiple cron schedules with a custom timetable class [OK]

Common Mistakes:

• Trying to put two cron expressions in one string
• Creating multiple DAGs instead of one with combined timetable
• Using sensors instead of proper timetable combination