Skip to main content

Airflow ⇄ Pulse

Intermediate ⏱ 15 min airflow · python · taskflow · ✅ ran E2E in repo

Airflow owns the DAG scheduling/retries; Pulse is the real-time stream-processing task it calls. Repo example: examples/airflow-pulse-bridge/ — validated end-to-end on Airflow 3.0.3 via dag.test() (in-process, no scheduler/webserver).

@dag order_scoring: make_order → score_via_pulse (@task, retries=2) → decide

dag.py is a TaskFlow DAG; the Pulse call is one @task making the universal two calls (POST .../x/order-scoring/in, GET .../events/order-scoring.score.out).

Run it — the one Airflow-specific trap​

Airflow is heavy and its legacy deps clash with a modern system Python — install it in a virtualenv with the official constraints:

pulse server start --dev
mkdir app && cp pulse.yaml app/ && (cd app && pulse deploy .)

python3 -m venv .venv && . .venv/bin/activate
PYV=$(python -c 'import sys;print(f"{sys.version_info.major}.{sys.version_info.minor}")')
pip install "apache-airflow==3.0.3" requests \
  --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-3.0.3/constraints-${PYV}.txt"

export AIRFLOW_HOME=$PWD/.airflow AIRFLOW__CORE__LOAD_EXAMPLES=False
airflow db migrate            # init the metadata DB once
python dag.py                 # runs dag_obj.test() end-to-end

Validated output​

O-2500 amount=2500 -> high_value=True decision=escalate

Airflow schedules and retries in minutes; Pulse holds per-key state in milliseconds. That's the division: the DAG stays batch-shaped, the stream intelligence lives in the stage you can enrich freely.

Next​