Pipeline Orchestration with a Scheduler

Pipeline Orchestration with a Scheduler

Once an ETL pipeline works correctly when run by hand, the next step is to run it automatically—on a schedule, without anyone sitting at a keyboard. This is called pipeline orchestration. In production database environments, ETL jobs commonly run overnight, hourly, or on some other fixed interval so that target tables stay fresh and analysts find current data when they arrive in the morning.

This lesson covers two scheduling tools every DBA should know: cron on Linux and Task Scheduler on Windows. It also introduces the pattern of writing a notification record to the database after each run, and provides a first look at Apache Airflow, the leading open-source orchestration platform.

How cron Works

cron is a time-based job scheduler built into Linux and macOS. Each user has a personal list of scheduled commands called a crontab (cron table). You edit it with `crontab -e` and list it with `crontab -l`.

Each line in a crontab has five time fields followed by the command:

```

• command-to-run

│ │ │ │ │

│ │ │ │ └─ day of week (0=Sunday … 6=Saturday)

│ │ │ └─── month (1-12)

│ │ └───── day of month (1-31)

│ └─────── hour (0-23)

└───────── minute (0-59)

```

To run a Python ETL script every night at 2:00 AM:

```

0 2 * /usr/bin/python3 /home/student/etl/run_pipeline.py >> /home/student/etl/etl.log 2>&1

```

The `>> ... 2>&1` portion appends both standard output and error messages to a log file. Always redirect output to a log file; without it, cron silently discards messages and troubleshooting becomes very difficult.

Important operating practice: Use full absolute paths for both the interpreter and the script. cron runs with a minimal environment and may not find commands that work fine in your interactive shell.

Windows Task Scheduler

On Windows, the equivalent tool is Task Scheduler. You create a task, set a trigger (daily, weekly, on a schedule), and point the action at the Python interpreter with the script path as an argument. Like cron, Task Scheduler can redirect output to a log file using a wrapper batch script. The underlying principle—run a command at a defined time without human intervention—is identical on both platforms.

Writing a Notifications Table

Scheduled jobs run unattended, so you need a reliable way to know whether a run succeeded or failed. A simple and portable pattern is to write a summary row to a notifications table in PostgreSQL at the end of every pipeline run. A minimal table might look like:

```sql

CREATE TABLE etl_notifications (

id SERIAL PRIMARY KEY,

pipeline TEXT NOT NULL,

status TEXT NOT NULL, -- 'SUCCESS' or 'FAILURE'

rows_loaded INTEGER,

run_at TIMESTAMPTZ DEFAULT now(),

message TEXT

);

```

At the end of your Python ETL script, insert a row:

```python

cur.execute(

"INSERT INTO etl_notifications (pipeline, status, rows_loaded, message) "

"VALUES (%s, %s, %s, %s)",

('nightly_sales', 'SUCCESS', row_count, 'Pipeline completed normally')

)

conn.commit()

```

If an exception is caught, insert a FAILURE row instead and include the error text in the message column. This gives you a durable, queryable audit trail—an approach used by commercial orchestrators such as Informatica and Azure Data Factory, scaled up with richer metadata.

Introduction to Apache Airflow

For complex pipelines with many interdependent steps, a dedicated orchestration platform adds significant value. Apache Airflow is a widely used open-source tool. In Airflow, you define a pipeline as a DAG (Directed Acyclic Graph)—a Python file that describes tasks and the order they must run. Key concepts:

• DAG: The overall pipeline definition, including its schedule interval.
• Task: A single unit of work (run a SQL query, call a Python function, trigger an API).
• Operator: The type of task—PythonOperator, PostgresOperator, BashOperator, and many others.
• Task dependency: Declared with `>>` syntax, e.g., `extract >> transform >> load`.
• Retry settings: Each task can be configured to retry on failure, with a delay between attempts.
• Logs: Airflow stores per-task logs accessible through its web UI.

The DAG structure makes dependencies explicit and visual, which reduces errors compared to maintaining complex cron entries manually. Everything you built with cron and the notifications table maps directly to Airflow concepts: the schedule is the DAG's `schedule_interval`, the log file becomes Airflow task logs, and the notifications table pattern is replaced by Airflow's built-in metadata database.

Safety and Professional Practices

Scheduled jobs run with whatever permissions the user account has. Follow the principle of least privilege: the database user your ETL script connects as should have only the permissions it needs—no superuser rights for a routine load job. Store database credentials in environment variables or a secrets manager, never hard-coded in the script.

Before activating a schedule in production, test the full run manually and confirm the log file and notifications table record look correct. Keep a retention policy for log files; unbounded log growth can fill a disk and halt the server. When changing a live crontab on a shared Linux system, verify with `crontab -l` immediately after editing so you can catch syntax errors before the next scheduled window.