Aller au contenu principal

Airflow ⇄ Pulse

Intermédiaire ⏱ 15 min airflow · python · taskflow · ✅ exécuté E2E dans le repo

Airflow possède le scheduling/retries du DAG ; Pulse est la task de traitement de flux temps réel qu'il appelle. Exemple du repo : examples/airflow-pulse-bridge/ — validé de bout en bout sur Airflow 3.0.3 via dag.test() (in-process, sans scheduler/webserver).

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

dag.py est un DAG TaskFlow ; l'appel Pulse est un seul @task qui fait les deux appels universels (POST .../x/order-scoring/in, GET .../events/order-scoring.score.out).

Lancez-le — le seul piège spécifique à Airflow​

Airflow est lourd et ses dépendances legacy cassent sur un Python système moderne — installez-le dans un virtualenv avec les constraints officielles :

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 de la metadata DB, une fois
python dag.py                 # exécute dag_obj.test() de bout en bout

Sortie validée​

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

Airflow planifie et réessaie en minutes ; Pulse tient l'état par clé en millisecondes. C'est la division : le DAG reste taillé batch, l'intelligence de flux vit dans le stage que vous enrichissez librement.

Ensuite​