Data Engineering Basics: Build a Pipeline with Airflow and Docker

McTaba runs coding and data bootcamps in Kenya. This guide is written to be useful whether or not you ever join one.

Imagine a till that exports a CSV file every evening. Someone opens it, fixes the spelling, copies the numbers into a sheet and sometimes forgets. A pipeline does those steps by itself at 10 PM, in the same order, and tells you when one fails. That is data engineering in one picture, and a skills step on our data science career map.

Airflow is a tool that runs such steps on a schedule. You describe the steps in Python as a DAG: a set of tasks with an order. Docker lets you run Airflow without installing it piece by piece on your laptop.

Task	What it does
pull	Finds today's sales export and checks it exists
clean	Fixes names, drops duplicates and bad rows, adds revenue
load	Writes the rows into a database, safely, even if run twice

Our Data Science to AI Engineering programme covers Docker, FastAPI and monitoring in phase 6, and real-time pipelines in phase 7.

Install Docker Desktop first. The official guide, Running Airflow in Docker, shows Airflow 3.3.2 (checked on 3 November 2026). It asks for at least 4 GB of memory for Docker, and about 8 GB on macOS. It describes this setup as a way to learn and explore, not for production.

mkdir airflow-shop && cd airflow-shop
curl -LfO 'https://airflow.apache.org/docs/apache-airflow/3.3.2/docker-compose.yaml'
mkdir -p ./dags ./logs ./plugins ./config ./data/incoming
echo "AIRFLOW_UID=50000" > .env
docker compose up airflow-init
docker compose up

The guide's own Linux command writes your user id into .env. On other systems, the number 50000 does the job. The version number sits inside the URL, so open the guide and copy its current line if it has moved on.

Open http://localhost:8080 and sign in with the user airflow and the password airflow, which the guide gives as the defaults. Never expose that login to the internet.

One edit is ours. Our DAG reads files from a data folder, so add this line under volumes: in the compose file, beside the dags line:

    - ${AIRFLOW_PROJ_DIR:-.}/data:/opt/airflow/data

Run docker compose down, then docker compose up again so the new mount takes effect.

Save this as dags/daily_sales.py. It uses the TaskFlow style from the Airflow tutorial, from airflow.sdk import dag, task. The sales files are ones you make up: a CSV named sales_2026-11-02.csv with the columns date, product, qty and unit_price, placed in data/incoming.

import sqlite3
from pathlib import Path

import pandas as pd
import pendulum
from airflow.sdk import dag, task

DATA = Path("/opt/airflow/data")


def day(ds):
    # a manual run may have no ds, so fall back to today in Nairobi
    return ds or pendulum.now("Africa/Nairobi").to_date_string()


@dag(
    schedule="0 22 * * *",
    start_date=pendulum.datetime(2026, 11, 1, tz="Africa/Nairobi"),
    catchup=False,
    tags=["shop"],
)
def daily_sales():
    @task
    def pull(ds=None):
        # the till exports sales_YYYY-MM-DD.csv into data/incoming
        ds = day(ds)
        src = DATA / "incoming" / f"sales_{ds}.csv"
        if not src.exists():
            raise FileNotFoundError(f"No export yet: {src}")
        return str(src)

    @task
    def clean(src, ds=None):
        ds = day(ds)
        df = pd.read_csv(src)
        df["product"] = df["product"].str.strip().str.capitalize()
        df = df.drop_duplicates().dropna(subset=["qty", "unit_price"])
        df = df[df["qty"] > 0]
        df["revenue"] = df["qty"] * df["unit_price"]
        out = DATA / f"clean_{ds}.csv"
        df.to_csv(out, index=False)
        return str(out)

    @task
    def load(path, ds=None):
        ds = day(ds)
        df = pd.read_csv(path)
        with sqlite3.connect(DATA / "shop.db") as con:
            con.execute(
                "CREATE TABLE IF NOT EXISTS sales ("
                "date TEXT, product TEXT, qty INT, "
                "unit_price REAL, revenue REAL)")
            con.execute("DELETE FROM sales WHERE date = ?", (ds,))
            df.to_sql("sales", con, if_exists="append", index=False)

    load(clean(pull()))


daily_sales()

Three design choices are worth copying. The pull task fails loudly when the file is missing, so a quiet day is not mistaken for a good one. The clean step writes a new file and never touches the original. The load step deletes that day's rows before inserting, so running it twice gives the same table. Engineers call that idempotent.

The schedule is a cron string, 0 22 * * *, meaning 10 PM every day in Nairobi time. We used SQLite to keep the example light. A real shop would load into PostgreSQL.

The DAG appears in the web page within a minute or two. Switch it on, press Trigger, and watch the three boxes turn green in the Graph view. If one turns red, click it and read its log. The first run usually fails on a typo in the file name or a missing mount.

Check the data with sqlite3 data/shop.db "select product, sum(revenue) from sales group by product". For what to do with the table, see our pandas tutorial on a shop's sales and forecasting a shop's stock from till data.

Before you trust it

• Test the cleaning code on a bad file by hand before you schedule it.
• Add retries and an alert, an email or a WhatsApp message, for failed runs.
• Keep secrets out of the DAG file. Use Airflow connections or environment variables.

When the model is ready to answer live requests, serve it with FastAPI in a container. If you prefer to query the loaded table in SQL, start with SQL for data analysis.

To build this skill with mentors, see our Data Science to AI Engineering programme: 5 months full time or 8 months part time, KES 140,000, starting 11 January 2027.