Infrastructure·CI/CD & Automation·Beginner

Airflow Data Pipeline

Build an Apache Airflow DAG for orchestrating complex end-to-end data and ML pipelines including multi-step data fetching, cleaning, feature engineering, model training, and prediction generation tasks with full inter-task dependency management, automatic retry, and failure alerting.

automationinfrastructuretechnical-analysis

Scheduling Data Pipelines with Airflow

[8]

# Install Apache Airflow to resolve ModuleNotFoundError for conceptual examples.
# Note: A full Airflow environment requires more extensive setup (database, scheduler, webserver)
# and is typically run outside a standard Colab notebook.
!pip install apache-airflow

Collecting apache-airflow
  Downloading apache_airflow-3.2.2-py3-none-any.whl.metadata (36 kB)
Collecting apache-airflow-core==3.2.2 (from apache-airflow)
  Downloading apache_airflow_core-3.2.2-py3-none-any.whl.metadata (6.7 kB)
Collecting apache-airflow-task-sdk==1.2.2 (from apache-airflow)
  Downloading apache_airflow_task_sdk-1.2.2-py3-none-any.whl.metadata (5.0 kB)
Collecting a2wsgi>=1.10.8 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading a2wsgi-1.10.10-py3-none-any.whl.metadata (4.0 kB)
Collecting aiosqlite<0.22.0,>=0.20.0 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading aiosqlite-0.21.0-py3-none-any.whl.metadata (4.3 kB)
Requirement already satisfied: alembic<2.0,>=1.13.1 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (1.18.4)
Collecting apache-airflow-providers-common-compat>=1.7.4 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading apache_airflow_providers_common_compat-1.15.0-py3-none-any.whl.metadata (5.9 kB)
Collecting apache-airflow-providers-common-io>=1.6.3 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading apache_airflow_providers_common_io-1.7.3-py3-none-any.whl.metadata (5.7 kB)
Collecting apache-airflow-providers-common-sql>=1.28.1 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading apache_airflow_providers_common_sql-2.0.1-py3-none-any.whl.metadata (8.0 kB)
Collecting apache-airflow-providers-smtp>=2.3.1 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading apache_airflow_providers_smtp-3.0.1-py3-none-any.whl.metadata (5.3 kB)
Collecting apache-airflow-providers-standard>=1.9.0 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading apache_airflow_providers_standard-1.14.0-py3-none-any.whl.metadata (5.7 kB)
Collecting argcomplete>=1.10 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading argcomplete-3.6.3-py3-none-any.whl.metadata (16 kB)
Collecting asgiref>=2.3.0 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading asgiref-3.11.1-py3-none-any.whl.metadata (9.3 kB)
Requirement already satisfied: attrs!=25.2.0,>=22.1.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (26.1.0)
Requirement already satisfied: cachetools>=6.0.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (6.2.6)
Collecting cadwyn>=6.1.1 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading cadwyn-7.0.0-py3-none-any.whl.metadata (4.5 kB)
Collecting colorlog>=6.8.2 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading colorlog-6.10.1-py3-none-any.whl.metadata (11 kB)
Collecting cron-descriptor>=1.2.24 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading cron_descriptor-2.1.0-py3-none-any.whl.metadata (8.1 kB)
Collecting croniter>=2.0.2 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading croniter-6.2.2-py3-none-any.whl.metadata (22 kB)
Requirement already satisfied: cryptography>=44.0.3 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (48.0.0)
Collecting deprecated>=1.2.13 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading deprecated-1.3.1-py2.py3-none-any.whl.metadata (5.9 kB)
Requirement already satisfied: dill>=0.2.2 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (0.3.8)
Requirement already satisfied: fastapi>=0.129.0 in /usr/local/lib/python3.12/dist-packages (from fastapi[standard-no-fastapi-cloud-cli]>=0.129.0->apache-airflow-core==3.2.2->apache-airflow) (0.136.3)
Requirement already satisfied: httpx>=0.25.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (0.28.1)
Requirement already satisfied: importlib-metadata>=7.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (8.7.1)
Requirement already satisfied: itsdangerous>=2.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (2.2.0)
Requirement already satisfied: jinja2>=3.1.5 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (3.1.6)
Requirement already satisfied: jsonschema>=4.19.1 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (4.26.0)
Collecting lazy-object-proxy>=1.2.0 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading lazy_object_proxy-1.12.0-cp312-cp312-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl.metadata (5.1 kB)
Collecting libcst>=1.8.2 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading libcst-1.8.6-cp312-cp312-manylinux_2_28_x86_64.whl.metadata (15 kB)
Requirement already satisfied: linkify-it-py>=2.0.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (2.1.0)
Collecting lockfile>=0.12.2 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading lockfile-0.12.2-py2.py3-none-any.whl.metadata (2.4 kB)
Collecting methodtools>=0.4.7 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading methodtools-0.4.7-py2.py3-none-any.whl.metadata (3.0 kB)
Collecting msgspec>=0.19.0 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading msgspec-0.21.1-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl.metadata (5.8 kB)
Requirement already satisfied: natsort>=8.4.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (8.4.0)
Requirement already satisfied: opentelemetry-api>=1.27.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (1.38.0)
Collecting opentelemetry-exporter-otlp>=1.27.0 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading opentelemetry_exporter_otlp-1.42.1-py3-none-any.whl.metadata (2.4 kB)
Requirement already satisfied: opentelemetry-proto<9999,>=1.27.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (1.38.0)
Requirement already satisfied: packaging>=25.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (26.2)
Collecting pathspec>=0.9.0 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading pathspec-1.1.1-py3-none-any.whl.metadata (14 kB)
Collecting pendulum>=3.1.0 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading pendulum-3.2.0-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl.metadata (6.8 kB)
Requirement already satisfied: pluggy>=1.5.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (1.6.0)
Requirement already satisfied: psutil>=5.8.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (5.9.5)
Requirement already satisfied: pydantic>=2.11.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (2.12.3)
Requirement already satisfied: pygments!=2.19.0,>=2.0.1 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (2.20.0)
Collecting pygtrie>=2.5.0 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading pygtrie-2.5.0-py3-none-any.whl.metadata (7.5 kB)
Requirement already satisfied: pyjwt>=2.11.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (2.13.0)
Collecting python-daemon>=3.0.0 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading python_daemon-3.1.2-py3-none-any.whl.metadata (4.8 kB)
Requirement already satisfied: python-dateutil>=2.7.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (2.9.0.post0)
Requirement already satisfied: python-slugify>=5.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (8.0.4)
Requirement already satisfied: pyyaml>=6.0.3 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (6.0.3)
Requirement already satisfied: requests<3,>=2.32.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (2.32.4)
Collecting rich-argparse>=1.0.0 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading rich_argparse-1.8.0-py3-none-any.whl.metadata (15 kB)
Requirement already satisfied: rich>=13.6.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (13.9.4)
Collecting setproctitle>=1.3.3 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading setproctitle-1.3.7-cp312-cp312-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl.metadata (10 kB)
Requirement already satisfied: sqlalchemy>=2.0.48 in /usr/local/lib/python3.12/dist-packages (from sqlalchemy[asyncio]>=2.0.48->apache-airflow-core==3.2.2->apache-airflow) (2.0.50)
Collecting starlette>=1.0.1 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading starlette-1.2.1-py3-none-any.whl.metadata (6.3 kB)
Collecting structlog>=25.4.0 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading structlog-26.1.0-py3-none-any.whl.metadata (9.7 kB)
Collecting svcs>=25.1.0 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading svcs-25.1.0-py3-none-any.whl.metadata (7.6 kB)
Requirement already satisfied: tabulate>=0.9.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (0.9.0)
Requirement already satisfied: tenacity>=8.3.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (9.1.4)
Requirement already satisfied: termcolor>=3.0.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (3.3.0)
Requirement already satisfied: typing-extensions>=4.14.1 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (4.15.0)
Collecting universal-pathlib>=0.3.8 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading universal_pathlib-0.3.10-py3-none-any.whl.metadata (39 kB)
Collecting uuid6>=2024.7.10 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading uuid6-2025.0.1-py3-none-any.whl.metadata (10 kB)
Requirement already satisfied: uvicorn>=0.37.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-core==3.2.2->apache-airflow) (0.49.0)
Requirement already satisfied: babel>=2.17.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-task-sdk==1.2.2->apache-airflow) (2.18.0)
Requirement already satisfied: fsspec>=2023.10.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-task-sdk==1.2.2->apache-airflow) (2025.3.0)
Collecting greenback>=1.2.1 (from apache-airflow-task-sdk==1.2.2->apache-airflow)
  Downloading greenback-1.3.0-py3-none-any.whl.metadata (9.8 kB)
Collecting psutil>=5.8.0 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading psutil-7.2.2-cp36-abi3-manylinux2010_x86_64.manylinux_2_12_x86_64.manylinux_2_28_x86_64.whl.metadata (22 kB)
Requirement already satisfied: Mako in /usr/local/lib/python3.12/dist-packages (from alembic<2.0,>=1.13.1->apache-airflow-core==3.2.2->apache-airflow) (1.3.12)
Requirement already satisfied: sqlparse>=0.5.1 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-providers-common-sql>=1.28.1->apache-airflow-core==3.2.2->apache-airflow) (0.5.5)
Requirement already satisfied: more-itertools>=9.0.0 in /usr/local/lib/python3.12/dist-packages (from apache-airflow-providers-common-sql>=1.28.1->apache-airflow-core==3.2.2->apache-airflow) (10.8.0)
Collecting aiosmtplib>=0.1.6 (from apache-airflow-providers-smtp>=2.3.1->apache-airflow-core==3.2.2->apache-airflow)
  Downloading aiosmtplib-5.1.1-py3-none-any.whl.metadata (3.6 kB)
Requirement already satisfied: typing-inspection>=0.4.0 in /usr/local/lib/python3.12/dist-packages (from cadwyn>=6.1.1->apache-airflow-core==3.2.2->apache-airflow) (0.4.2)
Requirement already satisfied: cffi>=2.0.0 in /usr/local/lib/python3.12/dist-packages (from cryptography>=44.0.3->apache-airflow-core==3.2.2->apache-airflow) (2.0.0)
Requirement already satisfied: wrapt<3,>=1.10 in /usr/local/lib/python3.12/dist-packages (from deprecated>=1.2.13->apache-airflow-core==3.2.2->apache-airflow) (2.2.1)
Requirement already satisfied: annotated-doc>=0.0.2 in /usr/local/lib/python3.12/dist-packages (from fastapi>=0.129.0->fastapi[standard-no-fastapi-cloud-cli]>=0.129.0->apache-airflow-core==3.2.2->apache-airflow) (0.0.4)
Collecting fastapi-cli>=0.0.8 (from fastapi-cli[standard-no-fastapi-cloud-cli]>=0.0.8; extra == "standard-no-fastapi-cloud-cli"->fastapi[standard-no-fastapi-cloud-cli]>=0.129.0->apache-airflow-core==3.2.2->apache-airflow)
  Downloading fastapi_cli-0.0.24-py3-none-any.whl.metadata (6.4 kB)
Requirement already satisfied: python-multipart>=0.0.18 in /usr/local/lib/python3.12/dist-packages (from fastapi[standard-no-fastapi-cloud-cli]>=0.129.0->apache-airflow-core==3.2.2->apache-airflow) (0.0.32)
Collecting email-validator>=2.0.0 (from fastapi[standard-no-fastapi-cloud-cli]>=0.129.0->apache-airflow-core==3.2.2->apache-airflow)
  Downloading email_validator-2.3.0-py3-none-any.whl.metadata (26 kB)
Requirement already satisfied: pydantic-settings>=2.0.0 in /usr/local/lib/python3.12/dist-packages (from fastapi[standard-no-fastapi-cloud-cli]>=0.129.0->apache-airflow-core==3.2.2->apache-airflow) (2.14.1)
Collecting pydantic-extra-types>=2.0.0 (from fastapi[standard-no-fastapi-cloud-cli]>=0.129.0->apache-airflow-core==3.2.2->apache-airflow)
  Downloading pydantic_extra_types-2.11.1-py3-none-any.whl.metadata (4.2 kB)
Requirement already satisfied: greenlet!=0.4.17 in /usr/local/lib/python3.12/dist-packages (from greenback>=1.2.1->apache-airflow-task-sdk==1.2.2->apache-airflow) (3.5.1)
Requirement already satisfied: sniffio in /usr/local/lib/python3.12/dist-packages (from greenback>=1.2.1->apache-airflow-task-sdk==1.2.2->apache-airflow) (1.3.1)
Collecting outcome (from greenback>=1.2.1->apache-airflow-task-sdk==1.2.2->apache-airflow)
  Downloading outcome-1.3.0.post0-py2.py3-none-any.whl.metadata (2.6 kB)
Requirement already satisfied: anyio in /usr/local/lib/python3.12/dist-packages (from httpx>=0.25.0->apache-airflow-core==3.2.2->apache-airflow) (4.13.0)
Requirement already satisfied: certifi in /usr/local/lib/python3.12/dist-packages (from httpx>=0.25.0->apache-airflow-core==3.2.2->apache-airflow) (2026.5.20)
Requirement already satisfied: httpcore==1.* in /usr/local/lib/python3.12/dist-packages (from httpx>=0.25.0->apache-airflow-core==3.2.2->apache-airflow) (1.0.9)
Requirement already satisfied: idna in /usr/local/lib/python3.12/dist-packages (from httpx>=0.25.0->apache-airflow-core==3.2.2->apache-airflow) (3.18)
Requirement already satisfied: h11>=0.16 in /usr/local/lib/python3.12/dist-packages (from httpcore==1.*->httpx>=0.25.0->apache-airflow-core==3.2.2->apache-airflow) (0.16.0)
Requirement already satisfied: zipp>=3.20 in /usr/local/lib/python3.12/dist-packages (from importlib-metadata>=7.0->apache-airflow-core==3.2.2->apache-airflow) (4.1.0)
Requirement already satisfied: MarkupSafe>=2.0 in /usr/local/lib/python3.12/dist-packages (from jinja2>=3.1.5->apache-airflow-core==3.2.2->apache-airflow) (3.0.3)
Requirement already satisfied: jsonschema-specifications>=2023.03.6 in /usr/local/lib/python3.12/dist-packages (from jsonschema>=4.19.1->apache-airflow-core==3.2.2->apache-airflow) (2025.9.1)
Requirement already satisfied: referencing>=0.28.4 in /usr/local/lib/python3.12/dist-packages (from jsonschema>=4.19.1->apache-airflow-core==3.2.2->apache-airflow) (0.37.0)
Requirement already satisfied: rpds-py>=0.25.0 in /usr/local/lib/python3.12/dist-packages (from jsonschema>=4.19.1->apache-airflow-core==3.2.2->apache-airflow) (2026.5.1)
Requirement already satisfied: uc-micro-py in /usr/local/lib/python3.12/dist-packages (from linkify-it-py>=2.0.0->apache-airflow-core==3.2.2->apache-airflow) (2.0.0)
Collecting wirerope>=0.4.7 (from methodtools>=0.4.7->apache-airflow-core==3.2.2->apache-airflow)
  Downloading wirerope-1.0.0-py2.py3-none-any.whl.metadata (3.3 kB)
Collecting opentelemetry-exporter-otlp-proto-grpc==1.42.1 (from opentelemetry-exporter-otlp>=1.27.0->apache-airflow-core==3.2.2->apache-airflow)
  Downloading opentelemetry_exporter_otlp_proto_grpc-1.42.1-py3-none-any.whl.metadata (2.6 kB)
Collecting opentelemetry-exporter-otlp-proto-http==1.42.1 (from opentelemetry-exporter-otlp>=1.27.0->apache-airflow-core==3.2.2->apache-airflow)
  Downloading opentelemetry_exporter_otlp_proto_http-1.42.1-py3-none-any.whl.metadata (2.4 kB)
Requirement already satisfied: googleapis-common-protos~=1.57 in /usr/local/lib/python3.12/dist-packages (from opentelemetry-exporter-otlp-proto-grpc==1.42.1->opentelemetry-exporter-otlp>=1.27.0->apache-airflow-core==3.2.2->apache-airflow) (1.75.0)
Requirement already satisfied: grpcio<2.0.0,>=1.63.2 in /usr/local/lib/python3.12/dist-packages (from opentelemetry-exporter-otlp-proto-grpc==1.42.1->opentelemetry-exporter-otlp>=1.27.0->apache-airflow-core==3.2.2->apache-airflow) (1.81.0)
Collecting opentelemetry-exporter-otlp-proto-common==1.42.1 (from opentelemetry-exporter-otlp-proto-grpc==1.42.1->opentelemetry-exporter-otlp>=1.27.0->apache-airflow-core==3.2.2->apache-airflow)
  Downloading opentelemetry_exporter_otlp_proto_common-1.42.1-py3-none-any.whl.metadata (1.8 kB)
Collecting opentelemetry-proto<9999,>=1.27.0 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading opentelemetry_proto-1.42.1-py3-none-any.whl.metadata (2.3 kB)
Collecting opentelemetry-sdk~=1.42.1 (from opentelemetry-exporter-otlp-proto-grpc==1.42.1->opentelemetry-exporter-otlp>=1.27.0->apache-airflow-core==3.2.2->apache-airflow)
  Downloading opentelemetry_sdk-1.42.1-py3-none-any.whl.metadata (1.7 kB)
Requirement already satisfied: protobuf<7.0,>=5.0 in /usr/local/lib/python3.12/dist-packages (from opentelemetry-proto<9999,>=1.27.0->apache-airflow-core==3.2.2->apache-airflow) (5.29.6)
Requirement already satisfied: tzdata>=2020.1 in /usr/local/lib/python3.12/dist-packages (from pendulum>=3.1.0->apache-airflow-core==3.2.2->apache-airflow) (2026.2)
Requirement already satisfied: annotated-types>=0.6.0 in /usr/local/lib/python3.12/dist-packages (from pydantic>=2.11.0->apache-airflow-core==3.2.2->apache-airflow) (0.7.0)
Requirement already satisfied: pydantic-core==2.41.4 in /usr/local/lib/python3.12/dist-packages (from pydantic>=2.11.0->apache-airflow-core==3.2.2->apache-airflow) (2.41.4)
Requirement already satisfied: six>=1.5 in /usr/local/lib/python3.12/dist-packages (from python-dateutil>=2.7.0->apache-airflow-core==3.2.2->apache-airflow) (1.17.0)
Requirement already satisfied: text-unidecode>=1.3 in /usr/local/lib/python3.12/dist-packages (from python-slugify>=5.0->apache-airflow-core==3.2.2->apache-airflow) (1.3)
Requirement already satisfied: charset_normalizer<4,>=2 in /usr/local/lib/python3.12/dist-packages (from requests<3,>=2.32.0->apache-airflow-core==3.2.2->apache-airflow) (3.4.7)
Requirement already satisfied: urllib3<3,>=1.21.1 in /usr/local/lib/python3.12/dist-packages (from requests<3,>=2.32.0->apache-airflow-core==3.2.2->apache-airflow) (2.5.0)
Requirement already satisfied: markdown-it-py>=2.2.0 in /usr/local/lib/python3.12/dist-packages (from rich>=13.6.0->apache-airflow-core==3.2.2->apache-airflow) (4.2.0)
Collecting pathlib-abc<0.6.0,>=0.5.1 (from universal-pathlib>=0.3.8->apache-airflow-core==3.2.2->apache-airflow)
  Downloading pathlib_abc-0.5.2-py3-none-any.whl.metadata (4.6 kB)
Requirement already satisfied: click>=7.0 in /usr/local/lib/python3.12/dist-packages (from uvicorn>=0.37.0->apache-airflow-core==3.2.2->apache-airflow) (8.4.1)
Requirement already satisfied: pycparser in /usr/local/lib/python3.12/dist-packages (from cffi>=2.0.0->cryptography>=44.0.3->apache-airflow-core==3.2.2->apache-airflow) (3.0)
Collecting dnspython>=2.0.0 (from email-validator>=2.0.0->fastapi[standard-no-fastapi-cloud-cli]>=0.129.0->apache-airflow-core==3.2.2->apache-airflow)
  Downloading dnspython-2.8.0-py3-none-any.whl.metadata (5.7 kB)
Requirement already satisfied: typer>=0.16.0 in /usr/local/lib/python3.12/dist-packages (from fastapi-cli>=0.0.8->fastapi-cli[standard-no-fastapi-cloud-cli]>=0.0.8; extra == "standard-no-fastapi-cloud-cli"->fastapi[standard-no-fastapi-cloud-cli]>=0.129.0->apache-airflow-core==3.2.2->apache-airflow) (0.25.1)
Collecting rich-toolkit>=0.14.8 (from fastapi-cli>=0.0.8->fastapi-cli[standard-no-fastapi-cloud-cli]>=0.0.8; extra == "standard-no-fastapi-cloud-cli"->fastapi[standard-no-fastapi-cloud-cli]>=0.129.0->apache-airflow-core==3.2.2->apache-airflow)
  Downloading rich_toolkit-0.20.1-py3-none-any.whl.metadata (1.0 kB)
Requirement already satisfied: mdurl~=0.1 in /usr/local/lib/python3.12/dist-packages (from markdown-it-py>=2.2.0->rich>=13.6.0->apache-airflow-core==3.2.2->apache-airflow) (0.1.2)
Requirement already satisfied: python-dotenv>=0.21.0 in /usr/local/lib/python3.12/dist-packages (from pydantic-settings>=2.0.0->fastapi[standard-no-fastapi-cloud-cli]>=0.129.0->apache-airflow-core==3.2.2->apache-airflow) (1.2.2)
Requirement already satisfied: httptools>=0.8.0 in /usr/local/lib/python3.12/dist-packages (from uvicorn[standard]>=0.12.0; extra == "standard-no-fastapi-cloud-cli"->fastapi[standard-no-fastapi-cloud-cli]>=0.129.0->apache-airflow-core==3.2.2->apache-airflow) (0.8.0)
Requirement already satisfied: uvloop>=0.15.1 in /usr/local/lib/python3.12/dist-packages (from uvicorn[standard]>=0.12.0; extra == "standard-no-fastapi-cloud-cli"->fastapi[standard-no-fastapi-cloud-cli]>=0.129.0->apache-airflow-core==3.2.2->apache-airflow) (0.22.1)
Requirement already satisfied: watchfiles>=0.20 in /usr/local/lib/python3.12/dist-packages (from uvicorn[standard]>=0.12.0; extra == "standard-no-fastapi-cloud-cli"->fastapi[standard-no-fastapi-cloud-cli]>=0.129.0->apache-airflow-core==3.2.2->apache-airflow) (1.2.0)
Requirement already satisfied: websockets>=10.4 in /usr/local/lib/python3.12/dist-packages (from uvicorn[standard]>=0.12.0; extra == "standard-no-fastapi-cloud-cli"->fastapi[standard-no-fastapi-cloud-cli]>=0.129.0->apache-airflow-core==3.2.2->apache-airflow) (15.0.1)
Collecting opentelemetry-api>=1.27.0 (from apache-airflow-core==3.2.2->apache-airflow)
  Downloading opentelemetry_api-1.42.1-py3-none-any.whl.metadata (1.4 kB)
Collecting opentelemetry-semantic-conventions==0.63b1 (from opentelemetry-sdk~=1.42.1->opentelemetry-exporter-otlp-proto-grpc==1.42.1->opentelemetry-exporter-otlp>=1.27.0->apache-airflow-core==3.2.2->apache-airflow)
  Downloading opentelemetry_semantic_conventions-0.63b1-py3-none-any.whl.metadata (2.4 kB)
Requirement already satisfied: shellingham>=1.3.0 in /usr/local/lib/python3.12/dist-packages (from typer>=0.16.0->fastapi-cli>=0.0.8->fastapi-cli[standard-no-fastapi-cloud-cli]>=0.0.8; extra == "standard-no-fastapi-cloud-cli"->fastapi[standard-no-fastapi-cloud-cli]>=0.129.0->apache-airflow-core==3.2.2->apache-airflow) (1.5.4)
Downloading apache_airflow-3.2.2-py3-none-any.whl (12 kB)
Downloading apache_airflow_core-3.2.2-py3-none-any.whl (6.1 MB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m6.1/6.1 MB[0m [31m65.0 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading apache_airflow_task_sdk-1.2.2-py3-none-any.whl (492 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m492.8/492.8 kB[0m [31m34.1 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading a2wsgi-1.10.10-py3-none-any.whl (17 kB)
Downloading aiosqlite-0.21.0-py3-none-any.whl (15 kB)
Downloading apache_airflow_providers_common_compat-1.15.0-py3-none-any.whl (42 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m42.4/42.4 kB[0m [31m3.3 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading apache_airflow_providers_common_io-1.7.3-py3-none-any.whl (19 kB)
Downloading apache_airflow_providers_common_sql-2.0.1-py3-none-any.whl (96 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m96.3/96.3 kB[0m [31m10.3 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading apache_airflow_providers_smtp-3.0.1-py3-none-any.whl (26 kB)
Downloading apache_airflow_providers_standard-1.14.0-py3-none-any.whl (160 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m160.4/160.4 kB[0m [31m14.9 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading argcomplete-3.6.3-py3-none-any.whl (43 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m43.8/43.8 kB[0m [31m3.7 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading asgiref-3.11.1-py3-none-any.whl (24 kB)
Downloading cadwyn-7.0.0-py3-none-any.whl (61 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m61.2/61.2 kB[0m [31m5.9 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading colorlog-6.10.1-py3-none-any.whl (11 kB)
Downloading cron_descriptor-2.1.0-py3-none-any.whl (74 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m74.7/74.7 kB[0m [31m6.6 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading croniter-6.2.2-py3-none-any.whl (45 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m45.4/45.4 kB[0m [31m4.0 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading deprecated-1.3.1-py2.py3-none-any.whl (11 kB)
Downloading greenback-1.3.0-py3-none-any.whl (29 kB)
Downloading lazy_object_proxy-1.12.0-cp312-cp312-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl (71 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m71.5/71.5 kB[0m [31m3.7 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading libcst-1.8.6-cp312-cp312-manylinux_2_28_x86_64.whl (2.3 MB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m2.3/2.3 MB[0m [31m90.8 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading lockfile-0.12.2-py2.py3-none-any.whl (13 kB)
Downloading methodtools-0.4.7-py2.py3-none-any.whl (4.0 kB)
Downloading msgspec-0.21.1-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl (225 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m225.0/225.0 kB[0m [31m17.3 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading opentelemetry_exporter_otlp-1.42.1-py3-none-any.whl (6.7 kB)
Downloading opentelemetry_exporter_otlp_proto_grpc-1.42.1-py3-none-any.whl (19 kB)
Downloading opentelemetry_proto-1.42.1-py3-none-any.whl (71 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m71.8/71.8 kB[0m [31m6.9 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading opentelemetry_exporter_otlp_proto_http-1.42.1-py3-none-any.whl (21 kB)
Downloading opentelemetry_exporter_otlp_proto_common-1.42.1-py3-none-any.whl (17 kB)
Downloading pathspec-1.1.1-py3-none-any.whl (57 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m57.3/57.3 kB[0m [31m5.1 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading pendulum-3.2.0-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (348 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m348.9/348.9 kB[0m [31m30.2 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading psutil-7.2.2-cp36-abi3-manylinux2010_x86_64.manylinux_2_12_x86_64.manylinux_2_28_x86_64.whl (155 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m155.6/155.6 kB[0m [31m15.2 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading pygtrie-2.5.0-py3-none-any.whl (25 kB)
Downloading python_daemon-3.1.2-py3-none-any.whl (30 kB)
Downloading rich_argparse-1.8.0-py3-none-any.whl (25 kB)
Downloading setproctitle-1.3.7-cp312-cp312-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl (32 kB)
Downloading starlette-1.2.1-py3-none-any.whl (73 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m73.3/73.3 kB[0m [31m5.2 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading structlog-26.1.0-py3-none-any.whl (73 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m73.8/73.8 kB[0m [31m6.0 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading svcs-25.1.0-py3-none-any.whl (19 kB)
Downloading universal_pathlib-0.3.10-py3-none-any.whl (83 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m83.5/83.5 kB[0m [31m7.3 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading uuid6-2025.0.1-py3-none-any.whl (7.0 kB)
Downloading aiosmtplib-5.1.1-py3-none-any.whl (27 kB)
Downloading email_validator-2.3.0-py3-none-any.whl (35 kB)
Downloading fastapi_cli-0.0.24-py3-none-any.whl (12 kB)
Downloading pathlib_abc-0.5.2-py3-none-any.whl (19 kB)
Downloading pydantic_extra_types-2.11.1-py3-none-any.whl (79 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m79.5/79.5 kB[0m [31m6.7 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading wirerope-1.0.0-py2.py3-none-any.whl (9.2 kB)
Downloading outcome-1.3.0.post0-py2.py3-none-any.whl (10 kB)
Downloading dnspython-2.8.0-py3-none-any.whl (331 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m331.1/331.1 kB[0m [31m28.1 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading opentelemetry_sdk-1.42.1-py3-none-any.whl (170 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m170.9/170.9 kB[0m [31m16.1 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading opentelemetry_api-1.42.1-py3-none-any.whl (61 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m61.3/61.3 kB[0m [31m5.6 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading opentelemetry_semantic_conventions-0.63b1-py3-none-any.whl (203 kB)
[2K   [90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━[0m [32m203.7/203.7 kB[0m [31m15.8 MB/s[0m eta [36m0:00:00[0m
[?25hDownloading rich_toolkit-0.20.1-py3-none-any.whl (35 kB)
Installing collected packages: pygtrie, lockfile, wirerope, uuid6, svcs, structlog, setproctitle, python-daemon, psutil, pathspec, pathlib-abc, outcome, opentelemetry-proto, opentelemetry-api, msgspec, libcst, lazy-object-proxy, dnspython, deprecated, cron-descriptor, colorlog, asgiref, argcomplete, aiosqlite, aiosmtplib, a2wsgi, universal-pathlib, starlette, pendulum, opentelemetry-semantic-conventions, opentelemetry-exporter-otlp-proto-common, methodtools, greenback, email-validator, croniter, rich-toolkit, rich-argparse, pydantic-extra-types, opentelemetry-sdk, opentelemetry-exporter-otlp-proto-http, opentelemetry-exporter-otlp-proto-grpc, fastapi-cli, cadwyn, opentelemetry-exporter-otlp, apache-airflow-providers-common-compat, apache-airflow-providers-standard, apache-airflow-providers-smtp, apache-airflow-providers-common-sql, apache-airflow-providers-common-io, apache-airflow-task-sdk, apache-airflow-core, apache-airflow
  Attempting uninstall: psutil
    Found existing installation: psutil 5.9.5
    Uninstalling psutil-5.9.5:
      Successfully uninstalled psutil-5.9.5
  Attempting uninstall: opentelemetry-proto
    Found existing installation: opentelemetry-proto 1.38.0
    Uninstalling opentelemetry-proto-1.38.0:
      Successfully uninstalled opentelemetry-proto-1.38.0
  Attempting uninstall: opentelemetry-api
    Found existing installation: opentelemetry-api 1.38.0
    Uninstalling opentelemetry-api-1.38.0:
      Successfully uninstalled opentelemetry-api-1.38.0
  Attempting uninstall: aiosqlite
    Found existing installation: aiosqlite 0.22.1
    Uninstalling aiosqlite-0.22.1:
      Successfully uninstalled aiosqlite-0.22.1
  Attempting uninstall: starlette
    Found existing installation: starlette 0.52.1
    Uninstalling starlette-0.52.1:
      Successfully uninstalled starlette-0.52.1
  Attempting uninstall: opentelemetry-semantic-conventions
    Found existing installation: opentelemetry-semantic-conventions 0.59b0
    Uninstalling opentelemetry-semantic-conventions-0.59b0:
      Successfully uninstalled opentelemetry-semantic-conventions-0.59b0
  Attempting uninstall: opentelemetry-exporter-otlp-proto-common
    Found existing installation: opentelemetry-exporter-otlp-proto-common 1.38.0
    Uninstalling opentelemetry-exporter-otlp-proto-common-1.38.0:
      Successfully uninstalled opentelemetry-exporter-otlp-proto-common-1.38.0
  Attempting uninstall: opentelemetry-sdk
    Found existing installation: opentelemetry-sdk 1.38.0
    Uninstalling opentelemetry-sdk-1.38.0:
      Successfully uninstalled opentelemetry-sdk-1.38.0
  Attempting uninstall: opentelemetry-exporter-otlp-proto-http
    Found existing installation: opentelemetry-exporter-otlp-proto-http 1.38.0
    Uninstalling opentelemetry-exporter-otlp-proto-http-1.38.0:
      Successfully uninstalled opentelemetry-exporter-otlp-proto-http-1.38.0
[31mERROR: pip's dependency resolver does not currently take into account all the packages that are installed. This behaviour is the source of the following dependency conflicts.
gradio 5.50.0 requires starlette<1.0,>=0.40.0, but you have starlette 1.2.1 which is incompatible.
opentelemetry-exporter-gcp-logging 1.11.0a0 requires opentelemetry-sdk<1.39.0,>=1.35.0, but you have opentelemetry-sdk 1.42.1 which is incompatible.
google-adk 1.29.0 requires opentelemetry-api<1.39.0,>=1.36.0, but you have opentelemetry-api 1.42.1 which is incompatible.
google-adk 1.29.0 requires opentelemetry-sdk<1.39.0,>=1.36.0, but you have opentelemetry-sdk 1.42.1 which is incompatible.
google-adk 1.29.0 requires starlette<1.0.0,>=0.49.1, but you have starlette 1.2.1 which is incompatible.[0m[31m
[0mSuccessfully installed a2wsgi-1.10.10 aiosmtplib-5.1.1 aiosqlite-0.21.0 apache-airflow-3.2.2 apache-airflow-core-3.2.2 apache-airflow-providers-common-compat-1.15.0 apache-airflow-providers-common-io-1.7.3 apache-airflow-providers-common-sql-2.0.1 apache-airflow-providers-smtp-3.0.1 apache-airflow-providers-standard-1.14.0 apache-airflow-task-sdk-1.2.2 argcomplete-3.6.3 asgiref-3.11.1 cadwyn-7.0.0 colorlog-6.10.1 cron-descriptor-2.1.0 croniter-6.2.2 deprecated-1.3.1 dnspython-2.8.0 email-validator-2.3.0 fastapi-cli-0.0.24 greenback-1.3.0 lazy-object-proxy-1.12.0 libcst-1.8.6 lockfile-0.12.2 methodtools-0.4.7 msgspec-0.21.1 opentelemetry-api-1.42.1 opentelemetry-exporter-otlp-1.42.1 opentelemetry-exporter-otlp-proto-common-1.42.1 opentelemetry-exporter-otlp-proto-grpc-1.42.1 opentelemetry-exporter-otlp-proto-http-1.42.1 opentelemetry-proto-1.42.1 opentelemetry-sdk-1.42.1 opentelemetry-semantic-conventions-0.63b1 outcome-1.3.0.post0 pathlib-abc-0.5.2 pathspec-1.1.1 pendulum-3.2.0 psutil-7.2.2 pydantic-extra-types-2.11.1 pygtrie-2.5.0 python-daemon-3.1.2 rich-argparse-1.8.0 rich-toolkit-0.20.1 setproctitle-1.3.7 starlette-1.2.1 structlog-26.1.0 svcs-25.1.0 universal-pathlib-0.3.10 uuid6-2025.0.1 wirerope-1.0.0

Note on Airflow Execution in Colab

The previous cells raised a ModuleNotFoundError because Apache Airflow was not installed in this Colab environment. The !pip install apache-airflow command above resolves this specific import error.

However, it's crucial to understand that even with the package installed, the Airflow DAGs defined in this notebook are conceptual examples for educational purposes. A full Airflow setup involves running a webserver, a scheduler, and configuring a backend database, which is typically done in a dedicated server environment or a Docker container, not directly within a standard Colab session. Therefore, these DAGs will not actively schedule or execute tasks in the background as they would in a functional Airflow deployment. The code demonstrates how you would define DAGs and tasks, but it won't actually orchestrate a live pipeline here.

Introduction to Apache Airflow

Apache Airflow is an open-source platform used to programmatically author, schedule, and monitor workflows. It allows data engineers and data scientists to create data pipelines as Directed Acyclic Graphs (DAGs) of tasks. This means you can define a sequence of operations that need to be executed, along with their dependencies, and Airflow will manage their execution and scheduling.

Why Airflow?

In modern data ecosystems, data pipelines are crucial for moving, transforming, and analyzing data. These pipelines often involve complex dependencies, require robust scheduling, and need clear monitoring capabilities. Airflow addresses these needs by providing:

Orchestration: Automates the execution of tasks in a defined order.
Scheduling: Allows pipelines to run at regular intervals (e.g., hourly, daily, weekly) or based on external triggers.
Monitoring: Provides a rich UI to visualize pipeline runs, check task status, and troubleshoot issues.
Scalability: Can be scaled to handle thousands of tasks across multiple workers.
Extensibility: Offers a wide range of operators and sensors for integrating with various data sources and services.

This notebook will guide you through the core concepts of Airflow, demonstrate how to define a simple pipeline, and illustrate key components.

Core Concepts of Airflow

Airflow relies on several fundamental concepts to define and manage workflows. Understanding these is key to building effective data pipelines.

1. Directed Acyclic Graphs (DAGs)

A DAG is a collection of tasks with dependencies, where no task can run until all its dependencies are met. The term "Directed" means tasks flow in one direction, and "Acyclic" means there are no loops, preventing tasks from running indefinitely. A DAG represents a complete workflow, from start to finish.

In Airflow, DAGs are defined in Python files. Each .py file typically contains one DAG object.

Key Characteristics of a DAG:

Start Date: The date from which the DAG starts scheduling runs.
Schedule Interval: How often the DAG runs (e.g., daily, hourly, cron expression).
Default Arguments: Common arguments for all tasks within the DAG (e.g., owner, retries).

Let's look at a conceptual example of how a DAG is structured in Python. (Note: This code won't run without an Airflow environment, but it demonstrates the structure).

[9]

from airflow import DAG
from datetime import datetime

# Although we won't run this, here's how a basic DAG definition looks.
# It's typically defined within a Python file in your Airflow DAGs folder.

# Define default arguments for the DAG
default_args = {
    'owner': 'airflow_user',
    'start_date': datetime(2023, 1, 1),
    'retries': 1,
}

# Instantiate a DAG object
# dag_id: Unique identifier for the DAG
# default_args: Dictionary of default arguments
# schedule: How often the DAG runs (e.g., '@daily', '0 0 * * *', 'None' for manual runs)
# catchup: Whether to run missed past schedules
# tags: Optional labels for organizing DAGs in the UI
with DAG(
    dag_id='my_first_airflow_dag',
    default_args=default_args,
    schedule='@daily', # Changed schedule_interval to schedule
    catchup=False,
    tags=['example', 'tutorial']
) as dag:
    # Tasks will be defined here later
    pass

print("Conceptual DAG 'my_first_airflow_dag' defined.")

Conceptual DAG 'my_first_airflow_dag' defined.

2. Operators

Operators are predefined templates for a single unit of work in Airflow. They encapsulate the logic for specific tasks, allowing you to quickly define what each step in your pipeline should do. Airflow comes with a rich set of built-in operators, and you can also create custom ones.

Common Operator Types:

BashOperator: Executes a bash command.
PythonOperator: Calls an arbitrary Python function.
PostgresOperator, MySqlOperator, S3Operator, BigQueryOperator: Interact with specific databases or cloud services.
DummyOperator: A placeholder operator, useful for structuring DAGs.

Each operator is an instance of a class, and when you instantiate it, you define its parameters (e.g., the bash command for a BashOperator, or the Python function for a PythonOperator).

[10]

from airflow import DAG
from airflow.providers.standard.operators.bash import BashOperator # Updated import
from airflow.providers.standard.operators.python import PythonOperator # Updated import
from datetime import datetime # Replaced days_ago import

# Define default arguments
default_args = {
    'owner': 'airflow_user',
    'start_date': datetime(2023, 1, 1), # Replaced days_ago(1) with a fixed datetime
    'retries': 1,
}

with DAG(
    dag_id='example_operators_dag',
    default_args=default_args,
    schedule=None, # Changed schedule_interval to schedule
    catchup=False,
    tags=['operators']
) as dag:
    # Task 1: BashOperator
    # Executes a simple bash command to print a message
    start_task = BashOperator(
        task_id='start_pipeline',
        bash_command='echo "Starting the data pipeline..."',
    )

    # Task 2: PythonOperator
    # Calls a Python function to perform some logic
    def greet_function(name):
        print(f"Hello, {name}! This is a Python task.")

    greet_task = PythonOperator(
        task_id='greet_user',
        python_callable=greet_function,
        op_kwargs={'name': 'Airflow User'},
    )

    # Define task dependencies using '>>' or '<<'
    # This means 'greet_task' will run after 'start_task'
    start_task >> greet_task

print("Conceptual DAG 'example_operators_dag' with Bash and Python operators defined.")

Conceptual DAG 'example_operators_dag' with Bash and Python operators defined.

3. Tasks and Task Instances

Task: An instantiation of an Operator within a DAG. When you create BashOperator(...) or PythonOperator(...) within a DAG context, you are defining a task.
Task Instance: A specific run of a task on a specific DAG run. For example, if your daily DAG has a task extract_data, then extract_data for January 1st, 2023, is a task instance, and extract_data for January 2nd, 2023, is another task instance. Task instances have states (e.g., running, success, failed, skipped).

4. Scheduling and Triggers

Airflow's core strength lies in its scheduling capabilities. You define how often a DAG should run using the schedule_interval parameter.

Schedule Interval Formats:

Cron Presets: @hourly, @daily, @weekly, @monthly, @yearly
Cron Expressions: Standard Unix cron syntax (e.g., '0 0 * * *' for daily at midnight).
Timedelta: A Python datetime.timedelta object for custom intervals.
None: For DAGs that are only triggered manually or by external events.

When a DAG runs according to its schedule, it creates a DAG Run, which then triggers the execution of all its task instances. Each DAG run has an execution_date which marks the start of its logical schedule period.

[11]

from airflow import DAG
from airflow.providers.standard.operators.empty import EmptyOperator # Changed import path for EmptyOperator
from datetime import datetime, timedelta # Replaced days_ago import

default_args = {
    'owner': 'airflow_user',
    'start_date': datetime(2023, 1, 1), # Replaced days_ago(2) with a fixed datetime
    'retries': 1,
}

# Example 1: Daily Schedule
with DAG(
    dag_id='daily_scheduled_dag',
    default_args=default_args,
    schedule='@daily', # Changed schedule_interval to schedule
    catchup=False,
    tags=['scheduling']
) as dag_daily:
    daily_task = EmptyOperator(
        task_id='run_daily_job',
    )

print("Conceptual Daily Scheduled DAG defined.")

# Example 2: Hourly Schedule
with DAG(
    dag_id='hourly_scheduled_dag',
    default_args=default_args,
    schedule='@hourly', # Changed schedule_interval to schedule
    catchup=False,
    tags=['scheduling']
) as dag_hourly:
    hourly_task = EmptyOperator(
        task_id='run_hourly_job',
    )

print("Conceptual Hourly Scheduled DAG defined.")

# Example 3: Custom Cron Schedule (e.g., every Monday at 9 AM)
with DAG(
    dag_id='custom_cron_dag',
    default_args=default_args,
    schedule='0 9 * * MON', # Changed schedule_interval to schedule
    catchup=False,
    tags=['scheduling']
) as dag_cron:
    cron_task = EmptyOperator(
        task_id='run_custom_cron_job',
    )

print("Conceptual Custom Cron Scheduled DAG defined.")

# Example 4: Timedelta Schedule (e.g., every 30 minutes)
with DAG(
    dag_id='timedelta_dag',
    default_args=default_args,
    schedule=timedelta(minutes=30), # Changed schedule_interval to schedule
    catchup=False,
    tags=['scheduling']
) as dag_timedelta:
    timedelta_task = EmptyOperator(
        task_id='run_timedelta_job',
    )

print("Conceptual Timedelta Scheduled DAG defined.")

Conceptual Daily Scheduled DAG defined.
Conceptual Hourly Scheduled DAG defined.
Conceptual Custom Cron Scheduled DAG defined.
Conceptual Timedelta Scheduled DAG defined.

Building a Simple Airflow Data Pipeline: ETL Example

Let's construct a conceptual End-to-End (ETL) data pipeline using Airflow, demonstrating how tasks are defined and chained together. This pipeline will simulate extracting data, transforming it, and then loading it into a destination.

Pipeline Steps:

Extract: Read raw data from a source.
Transform: Process and clean the raw data.
Load: Write the processed data to a target system.

[13]

import pandas as pd
import numpy as np

def extract_data(**kwargs):
    """
    Simulates extracting raw data. Generates a mock DataFrame.

    Inputs:
        **kwargs: Airflow context arguments (not directly used in this mock, but common).

    Outputs:
        pd.DataFrame: A DataFrame containing mock raw data.

    Formula:
        Generates a DataFrame with 'id', 'value', and 'category' columns.
    """
    print("Extracting data...")
    data = {
        'id': range(5),
        'value': np.random.randint(10, 100, 5),
        'category': np.random.choice(['A', 'B', 'C'], 5)
    }
    df = pd.DataFrame(data)
    print(f"Extracted {len(df)} rows.\n{df.head()}")
    # In a real Airflow environment, you'd push this data to XComs or a shared storage.
    # For this mock, we'll just return it conceptually.
    return df

def transform_data(ti, **kwargs):
    """
    Simulates transforming raw data. Multiplies 'value' by 2.

    Inputs:
        ti: TaskInstance object to pull data from XComs (conceptual here).
        **kwargs: Airflow context arguments.

    Outputs:
        pd.DataFrame: A DataFrame containing transformed data.

    Formula:
        `transformed_value = original_value * 2`
    """
    print("Transforming data...")
    # In Airflow, you'd pull data from a previous task using ti.xcom_pull
    # For this mock, we'll assume `extract_data` returned a DataFrame.
    # Let's create a sample if ti.xcom_pull is not truly working.
    raw_df = extract_data()

    transformed_df = raw_df.copy()
    transformed_df['value'] = transformed_df['value'] * 2
    print(f"Transformed {len(transformed_df)} rows.\n{transformed_df.head()}")
    return transformed_df

def load_data(ti, **kwargs):
    """
    Simulates loading transformed data into a destination.

    Inputs:
        ti: TaskInstance object to pull data from XComs (conceptual here).
        **kwargs: Airflow context arguments.

    Outputs:
        None.

    Formula:
        Simply prints the data that would be loaded.
    """
    print("Loading data...")
    # Again, assume `transform_data` returned a DataFrame conceptually.
    transformed_df = transform_data(None) # Pass None as ti for this mock

    # Simulate writing to a database or data lake
    print(f"Successfully loaded {len(transformed_df)} rows to destination.\nFinal data snapshot:\n{transformed_df.head()}")

print("Mock ETL functions defined.")

2026-06-10T11:31:08.994003Z [info     ] NumExpr defaulting to 2 threads. [numexpr.utils] loc=utils.py:164
Mock ETL functions defined.

Now, let's define the Airflow DAG that orchestrates these ETL functions using PythonOperator.

[15]

from airflow import DAG
from airflow.providers.standard.operators.python import PythonOperator # Updated import
from datetime import datetime, timedelta # Replaced days_ago import

default_args = {
    'owner': 'airflow_etl',
    'start_date': datetime(2023, 1, 1), # Replaced days_ago(1) with a fixed datetime
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    dag_id='simple_etl_pipeline',
    default_args=default_args,
    schedule=timedelta(hours=1), # Changed schedule_interval to schedule
    catchup=False,
    tags=['etl', 'data_pipeline']
) as etl_dag:
    # Define the extract task
    extract_task = PythonOperator(
        task_id='extract_data_task',
        python_callable=extract_data,
    )

    # Define the transform task
    transform_task = PythonOperator(
        task_id='transform_data_task',
        python_callable=transform_data,
    )

    # Define the load task
    load_task = PythonOperator(
        task_id='load_data_task',
        python_callable=load_data,
    )

    # Define the task dependencies
    # The 'extract_task' must complete successfully before 'transform_task' starts.
    # The 'transform_task' must complete successfully before 'load_task' starts.
    extract_task >> transform_task >> load_task

print("Conceptual 'simple_etl_pipeline' DAG defined with dependencies.")
print("\nThis DAG represents an hourly ETL process. The '>>' operator defines the order of execution.")
print("When this DAG is deployed in an Airflow environment, it will be automatically scheduled and executed.")

Conceptual 'simple_etl_pipeline' DAG defined with dependencies.

This DAG represents an hourly ETL process. The '>>' operator defines the order of execution.
When this DAG is deployed in an Airflow environment, it will be automatically scheduled and executed.

Visualizations of Data Pipelines

Airflow's web UI provides excellent visualizations of DAGs, showing their structure, task statuses, and execution history. While we cannot replicate the full Airflow UI here, we can create conceptual visualizations to understand pipeline flow and scheduling aspects using matplotlib and networkx.

1. Conceptual DAG Structure Visualization

This visualization uses networkx to represent the tasks and their dependencies as a graph. It helps in understanding the flow of data and operations within the pipeline. Each node is a task, and an arrow represents a dependency (one task must finish before the next can start).

[6]

import networkx as nx
import matplotlib.pyplot as plt

# Define the tasks and their dependencies for our ETL pipeline conceptually
# Format: (source_task, target_task)
dependencies = [
    ('extract_data_task', 'transform_data_task'),
    ('transform_data_task', 'load_data_task')
]

# Create a directed graph
G = nx.DiGraph(dependencies)

# Add nodes for tasks that might not have explicit dependencies (e.g., start/end markers)
all_tasks = set()
for u, v in dependencies:
    all_tasks.add(u)
    all_tasks.add(v)

# Add any isolated tasks if necessary, though in our ETL they're all connected
# For simplicity, we assume all tasks are covered by the dependencies list.

plt.figure(figsize=(10, 6))

# Position nodes using a layout algorithm
pos = nx.spring_layout(G, k=0.5, iterations=50) # Fruchterman-Reingold force-directed algorithm

# Draw nodes
nx.draw_networkx_nodes(G, pos, node_color='skyblue', node_size=3000)

# Draw edges (arrows)
nx.draw_networkx_edges(G, pos, edge_color='gray', arrows=True, arrowsize=20, width=2)

# Draw labels
nx.draw_networkx_labels(G, pos, font_size=10, font_weight='bold')

plt.title('Conceptual ETL Pipeline DAG Structure', size=15)
plt.axis('off') # Hide axes
plt.show()

print("\nInterpretation: This graph visually represents the workflow of our ETL pipeline. 'extract_data_task' is the starting point, followed by 'transform_data_task', and finally 'load_data_task'. The arrows clearly indicate the direction of flow and dependencies, meaning a downstream task will only begin once its upstream dependencies are successfully completed.")


Interpretation: This graph visually represents the workflow of our ETL pipeline. 'extract_data_task' is the starting point, followed by 'transform_data_task', and finally 'load_data_task'. The arrows clearly indicate the direction of flow and dependencies, meaning a downstream task will only begin once its upstream dependencies are successfully completed.

2. Conceptual Task Execution Timeline (Gantt-like Chart)

This visualization simulates a single run of our ETL pipeline over time, showing when each task starts and ends. This helps to understand task duration and potential parallelism (though our current ETL is sequential, a more complex DAG could have parallel tasks).

[7]

import matplotlib.pyplot as plt
import numpy as np

# Define tasks and their simulated durations (in minutes)
tasks = {
    'extract_data_task': {'duration': 10, 'start_offset': 0},
    'transform_data_task': {'duration': 15, 'start_offset': 10},
    'load_data_task': {'duration': 5, 'start_offset': 25} # starts after transform
}

task_names = list(tasks.keys())
task_durations = [tasks[name]['duration'] for name in task_names]
task_starts = [tasks[name]['start_offset'] for name in task_names]

fig, ax = plt.subplots(figsize=(12, 6))

# Create horizontal bars for each task
# Y-axis will be task names, X-axis will be time
ax.barh(task_names, task_durations, left=task_starts, color=['skyblue', 'lightcoral', 'lightgreen'])

# Add labels and title
ax.set_xlabel('Time (minutes)', fontsize=12)
ax.set_ylabel('Tasks', fontsize=12)
ax.set_title('Conceptual ETL Pipeline Execution Timeline (Single Run)', fontsize=14)

# Set x-axis limits to show the full timeline
max_time = max(s + d for s, d in zip(task_starts, task_durations))
ax.set_xlim(0, max_time + 5)

# Invert y-axis to have the first task at the top
ax.invert_yaxis()

# Add text labels for start and end times on each bar
for i, task_name in enumerate(task_names):
    start = task_starts[i]
    duration = task_durations[i]
    end = start + duration
    ax.text(start + duration / 2, i, f'{duration} min', va='center', ha='center', color='black', fontsize=9)
    ax.text(start, i - 0.2, f'Start: {start} min', va='top', ha='left', color='dimgray', fontsize=8)
    ax.text(end, i - 0.2, f'End: {end} min', va='top', ha='right', color='dimgray', fontsize=8)

plt.grid(axis='x', linestyle='--', alpha=0.7)
plt.tight_layout()
plt.show()

print("\nInterpretation: This timeline visualizes a simulated run of the ETL pipeline. It shows that 'extract_data_task' runs for 10 minutes, followed immediately by 'transform_data_task' for 15 minutes, and finally 'load_data_task' for 5 minutes. The entire pipeline run takes 30 minutes. This type of visualization helps in identifying bottlenecks, estimating overall pipeline duration, and understanding sequential execution.")


Interpretation: This timeline visualizes a simulated run of the ETL pipeline. It shows that 'extract_data_task' runs for 10 minutes, followed immediately by 'transform_data_task' for 15 minutes, and finally 'load_data_task' for 5 minutes. The entire pipeline run takes 30 minutes. This type of visualization helps in identifying bottlenecks, estimating overall pipeline duration, and understanding sequential execution.

Advanced Airflow Concepts (Brief Overview)

Airflow offers many powerful features for complex workflows beyond simple sequential tasks.

Sensors

Sensors are a special type of operator that waits for a certain condition to be met before proceeding. They are crucial for event-driven pipelines. Examples include:

FileSensor: Waits for a file to land in a specific location.
SqlSensor: Waits for a specific record to appear in a database table.
ExternalTaskSensor: Waits for a task in another DAG to complete.

Sensors prevent downstream tasks from running until their prerequisites (often external to Airflow) are satisfied.

Branching

Branching allows your DAG to follow different paths based on conditions (e.g., success/failure of a task, data values). The BranchPythonOperator is commonly used for this, where a Python function returns the task_id of the next task to execute.

This enables dynamic workflows where the execution path adapts to runtime conditions.

XComs (Cross-communication)

XComs (Cross-communication) are a mechanism for tasks to exchange small amounts of data. When a task returns a value, Airflow can automatically push that value into XComs. Downstream tasks can then pull these values to use them in their logic.

This is useful for passing small metadata, file paths, or configuration parameters between tasks.

Conclusion

Apache Airflow is a powerful and flexible platform for orchestrating and scheduling complex data pipelines. By understanding its core concepts—DAGs, Operators, Tasks, and Scheduling—you can design robust and automated workflows for various data engineering and machine learning applications.

This notebook provided a foundational understanding of how to define and conceptually visualize an Airflow data pipeline. To fully leverage Airflow, setting up an actual Airflow environment and experimenting with its web UI, different operators, and scheduling options is recommended.