Airflow pipeline example

Run bundle exec ruby -Ilib scripts/23_airflow.rb from an empty working directory. It creates generate/ there. Set AIRFLOW_DEMO_OTEL=1 to connect the existing collector and Jaeger. The generated project works without Mobilis afterward.

The airflow("orchestrator") DSL declares one logical service. Connect it to one postgres metadata/data store and one redis broker. It realizes into an API server, scheduler, DAG processor, Celery worker, triggerer, and a one-shot initialization service. The init service migrates the Airflow database and creates the local admin/admin account. Worker and scheduler restarts do not run initialization again. The PostgreSQL node retains its existing durable data mount; a Compose volume holds Airflow task logs.

service.local_package("mobilis-domain") { |pkg| pkg.write_file("src/mobilis_domain/__init__.py", source) } creates a standard src layout, pyproject.toml, and pytest suite. pkg.depends_on("requirement") declares Python dependencies. service.use_local_package(package) shares a package object with another Airflow or FastAPI declaration. Docker installs the distributions into the Airflow image using the official Airflow 3.1.0 Python 3.12 constraints. No source path injection is used. To edit a package, change generate/packages/NAME, then run ./dc_dev build and ./dc_dev up -d; DAG files under generate/orchestrator/dags are mounted for direct edits. To test packages outside Airflow, create a Python 3.12 venv, run pip install -e 'packages/mobilis-domain[test]' -e 'packages/mobilis-pipeline[test]', then run pytest packages/mobilis-domain/tests packages/mobilis-pipeline/tests from generate/.

From generate/, run ./dc_dev up -d. Open the port in development.env under ORCHESTRATOR_WEB_PORT; login is admin/admin. Trigger demo_ingest, wait for success, then demo_aggregate, then demo_report using the Airflow UI or ./dc_dev exec orchestrator airflow dags trigger DAG_ID. The DAGs are intentionally manual; declaration order alone does not impose cross-DAG dependencies. demo_ingest branches into parallel east and west extraction tasks, rejoins for a retry-safe upsert, and intentionally fails its first load attempt to demonstrate retry. The exact persisted aggregates are EAST: count 2, total 10.00 and WEST: count 2, total 30.25. demo_report stores JSON with grand total 40.25 in demo_reports. Repeat all three runs to check idempotence. Trigger demo_report with --conf '{"fail":true}' to see an intentional failed task.

Inspect results with ./dc_dev exec pipeline-db psql -U pipeline-db-development-user -d pipeline-db_development -c 'select * from demo_reports'. The generated development.env supplies credentials and database name if customized. Package imports can be checked using ./dc_dev exec orchestrator-worker python -c 'import mobilis_domain, mobilis_pipeline'. Run ./dc_dev exec orchestrator-worker python /opt/airflow/validate_dags.py for import, discovery and dependency assertions in the pinned Airflow image.

With the collector connection, Airflow native OTel traces and metrics use the connected collector endpoint. The generated packages read Airflow's task context carrier and attach its W3C trace context before creating domain, pipeline, and SQL spans. Jaeger then shows scheduler task spans and worker application spans in the same trace: load is the parent of pipeline.persist_sales, and that span is the parent of postgresql.insert. Airflow represents the task-to-DAG relationship as FOLLOWS_FROM, so it is a link rather than parentage. Component service names distinguish scheduler and worker. Inspect Jaeger at the generated JAEGER_* port and check span attributes such as airflow.dag_id, airflow.run_id, airflow.task_id, airflow.task_attempt, db.operation, and db.statement. After running ingest, aggregate, report, and the intentional failure with telemetry enabled, run python3 scripts/airflow/verify_traces.py from the Mobilis source checkout (pass the Jaeger URL as an argument if its port differs). The verifier checks actual exported ancestry, retry attributes, SQL spans, and the exception span. Without the collector connection, no exporter is configured and the pipeline still runs.