What You'll Build
In this exercise you will build a complete Apache Airflow DAG that orchestrates an end-to-end ML pipeline for IPL player churn prediction — identifying which players are at risk of declining performance or leaving the squad. The pipeline covers five production tasks in sequence: ingest_ipl_data ingests raw IPL batting statistics from a CSV file, feature_engineering computes predictive features using the cricket_feature_pipeline function, train_model fits a GradientBoostingClassifier on the engineered features, register_model logs the trained model to MLflow and registers it in the Model Registry under the champion slot, and notify_team sends a Slack-style notification with the training results. You will wire all five tasks using the Airflow TaskFlow API and the >> operator on a daily schedule.
Analogy🏏Cricket
🏏 Think of it like cricket: Before the IPL auction, a franchise's analytics team evaluates every player across dozens of trial matches — tracking Rohit Sharma's strike rate in power plays, Virat Kohli's average against pace, MS Dhoni's finishing rate in the death overs, Shubman Gill's consistency across pitches, and Jasprit Bumrah's economy in the middle overs. Each trial is recorded in a shared logbook so the selectors can compare and pick the best combination. MLflow is exactly that shared logbook for your ML experiments — every training run is a trial match, every metric is a scorecard entry, and the Model Registry is the final squad announcement. Keep the auction framing in mind throughout the exercise, because it fixes the discipline the steps teach: a franchise never signs a player off one good highlight reel, and you never register a model off one lucky run — you log every trial, compare them on identical conditions, and promote only with the full scorecard in front of you.
Prerequisites
Before starting, ensure you have Python 3.9 or later, Apache Airflow 2.5 or newer, and pip available. You will install four additional packages: scikit-learn for the GradientBoostingClassifier, mlflow for experiment tracking and model registration, pandas and numpy for data manipulation, and requests for the Slack webhook notification. Airflow must be initialised with airflow db init and at least one webserver and scheduler process running. Familiarity with Airflow's TaskFlow API decorator pattern from earlier reading lessons is assumed. No external cloud credentials are required — MLflow runs locally and the Slack notification uses a stub that prints to the console.
Analogy🏏Cricket
🏏 Think of it like cricket: the kit check before a net session. Just as a batter arriving for practice needs pads, gloves, and a bat they already know how to use — but doesn't need to have faced the new bowling machine before, because today's session is exactly where they'll learn it — this exercise expects you to arrive comfortable with Python, basic scikit-learn (fit, predict, train_test_split), and pandas, while MLflow itself is taught from scratch. Just as the coach insists on a properly prepared practice pitch — a clean, dedicated strip rather than the match square — you need a clean virtual environment or Conda environment where packages can be installed freely. And just as the only outside help needed is the equipment delivery van arriving once before practice, network access is required only for the initial pip install; after that everything runs locally, whether your 'net' is a laptop, a Docker container, or a cloud notebook. The payoff: checking your kit now means the session ahead is pure skill-building, with no stoppages for missing gear.
Setup
python
# Step 0: Install dependencies and create project structure
# Run in terminal inside your virtualenv:
# pip install apache-airflow==2.8.1 scikit-learn mlflow pandas numpy requests
# Verify key imports
import airflow
import mlflow
import sklearn
import pandas as pd
import numpy as np
print(f"airflow version: {airflow.__version__}")
print(f"mlflow version: {mlflow.__version__}")
print(f"sklearn version: {sklearn.__version__}")
print(f"pandas version: {pd.__version__}")
print("All dependencies verified.")
# Create project directories
import os
os.makedirs('/tmp/ipl_pipeline/data', exist_ok=True)
os.makedirs('/tmp/ipl_pipeline/models', exist_ok=True)
os.makedirs('/tmp/ipl_pipeline/staging', exist_ok=True)
# Generate sample IPL player dataset for the exercise
np.random.seed(42)
n_players = 200
ipl_players_dag = pd.DataFrame({
'match_id': [f'IPL2024_{i:04d}' for i in range(n_players)],
'player_name': np.random.choice(
['Rohit Sharma', 'Virat Kohli', 'Shubman Gill',
'MS Dhoni', 'Jasprit Bumrah', 'KL Rahul',
'Hardik Pandya', 'Suryakumar Yadav'],
n_players
),
'batting_average': np.random.normal(35.0, 12.0, n_players).clip(0, 80),
'innings_count': np.random.randint(5, 30, n_players),
'strike_rate_raw': np.random.normal(130.0, 20.0, n_players).clip(70, 220),
'boundaries_hit': np.random.randint(5, 80, n_players),
'total_runs': np.random.randint(80, 700, n_players),
'matches_played': np.random.randint(10, 50, n_players),
# Label: 1 = at risk of performance churn
'performance_churn': np.random.binomial(1, 0.35, n_players),
})
ipl_players_dag.to_csv('/tmp/ipl_pipeline/data/ipl_players_raw.csv', index=False)
print(f"Sample dataset: {ipl_players_dag.shape} saved to /tmp/ipl_pipeline/data/ipl_players_raw.csv")
print(ipl_players_dag.head(3).to_string())
Step 1: Define the DAG and Tasks
The DAG scaffolding defines the metadata, schedule, and default arguments that govern all five tasks. Using the @dag decorator from Airflow's TaskFlow API keeps the pipeline code readable and testable — each @task-decorated function becomes an Airflow task with automatic XCom serialization of its return value. The daily schedule ensures the pipeline refreshes the model every 24 hours with new match data. Setting catchup=False prevents Airflow from backfilling all missed days when the DAG is first enabled, which would create hundreds of redundant training runs. Setting retries=1 with a 5-minute retry delay handles transient failures like temporary file system unavailability without requiring manual re-triggers.
Analogy🏏Cricket
🏏 Think of it like cricket: the DAG definition is the official match fixture schedule — it specifies which teams play (tasks), on what dates (schedule_interval), the rules governing the match (default_args), and whether postponed matches need to be replayed (catchup). The @dag decorator is like the IPL Governing Council's formal fixture registration: once registered, every team (task) knows their role, their reporting time, and the protocol if something goes wrong (retries). Without this registration, no match happens regardless of how prepared the players are. Two details in the registration carry real weight: catchup=False tells the council not to schedule replays for all the historical dates before the fixture list existed — otherwise a newly deployed DAG tries to 'catch up' on months of imaginary matches — and default_args like retries with a delay are the standing rule that a rained-off session gets one more attempt after a break before anyone escalates.
python
# dags/ipl_churn_pipeline.py — DAG scaffold and shared constants
from __future__ import annotations
from datetime import datetime, timedelta
from airflow.decorators import dag, task
import os
# ── Shared path constants used by all five tasks ──────────────────────────────
RAW_DATA_PATH = '/tmp/ipl_pipeline/data/ipl_players_raw.csv'
STAGING_JSON = '/tmp/ipl_pipeline/staging/ipl_players_staging.json'
FEATURES_PARQUET = '/tmp/ipl_pipeline/staging/ipl_features.parquet'
METRICS_JSON = '/tmp/ipl_pipeline/staging/training_metrics.json'
MODEL_DIR = '/tmp/ipl_pipeline/models'
MLFLOW_TRACKING = 'sqlite:////tmp/ipl_pipeline/mlflow.db' # local SQLite backend
MLFLOW_MODEL_NAME = 'ipl_player_churn_model'
# ── Default arguments applied to all tasks ───────────────────────────────────
default_args = {
'owner': 'mlops_team',
'depends_on_past': False,
'email_on_failure': False,
'email_on_retry': False,
'retries': 1,
'retry_delay': timedelta(minutes=5),
}
@dag(
dag_id='ipl_player_churn_pipeline',
description='Daily IPL player churn prediction pipeline: '
'ingest → features → train → register → notify',
schedule_interval='@daily',
start_date=datetime(2024, 1, 1),
catchup=False, # never backfill missed runs
default_args=default_args,
tags=['mlops', 'cricket', 'churn', 'classification'],
)
def ipl_players_dag():
"""
End-to-end Airflow ML pipeline for IPL player performance churn prediction.
Task graph:
ingest_ipl_data
>> feature_engineering
>> train_model
>> register_model
>> notify_team
"""
# ── Task definitions live inside the @dag function (TaskFlow pattern) ─────
# Each task is defined as a nested @task-decorated function.
# TaskFlow automatically handles XCom push/pull via function return values.
pass # Task bodies defined in Steps 2-5 below
# Instantiate DAG (Airflow discovers this object)
ipl_churn_pipeline = ipl_players_dag()
Step 2: Data Ingestion Task
The ingest_ipl_data task is the pipeline's entry point. It reads raw IPL player batting statistics from the CSV file, performs lightweight structural validation to confirm mandatory columns are present and the file is non-empty, and writes the DataFrame to a staging JSON file that downstream tasks can consume. Keeping ingestion separate from validation and feature engineering follows the single-responsibility principle — if the data source changes from CSV to S3, only this task needs modification. The task returns the path to the staging file, which Airflow's XCom system passes automatically to the next task in the chain.
Analogy🏏Cricket
🏏 Think of it like cricket: ingest_ipl_data is MS Dhoni pulling the scorecards from last night's matches off the printer before the 6 AM team meeting. He does a quick structural check — are all match IDs present, are the innings counts filled in, is anything obviously wrong? If a scorecard is missing, he pages the scorers immediately rather than proceeding with incomplete data. The same fail-fast approach protects your pipeline: a missing required column raises an immediate exception rather than causing a confusing error 30 minutes later during model training. Dhoni also leaves the scorecards untouched — his job is to fetch and sanity-check, not to recalculate anyone's average; that separation keeps blame traceable when a number is wrong. And rather than reading the stack aloud to the next person, he leaves the checked pile in a labelled tray (the staging JSON file), with only the tray's label passed on via XCom — small notes travel between tasks, bulky data stays on disk.
python
# dags/ipl_churn_pipeline.py — Task 1: ingest_ipl_data
# (nested inside the @dag function body)
@task(task_id='ingest_ipl_data')
def ingest_ipl_data() -> str:
"""
Read raw IPL player statistics CSV; validate structure; write to staging JSON.
Returns: path to staging JSON file for downstream tasks.
"""
import pandas as pd
import os
if not os.path.exists(RAW_DATA_PATH):
raise FileNotFoundError(
f"Raw IPL data not found at {RAW_DATA_PATH}. "
"Ensure the upstream data pipeline has deposited the file."
)
ipl_players_dag_df = pd.read_csv(RAW_DATA_PATH)
# Structural validation — fail fast before any processing
required_columns = [
'match_id', 'player_name', 'batting_average',
'innings_count', 'strike_rate_raw', 'boundaries_hit',
'total_runs', 'matches_played', 'performance_churn'
]
missing_cols = [c for c in required_columns if c not in ipl_players_dag_df.columns]
if missing_cols:
raise ValueError(
f"Ingested IPL data is missing required columns: {missing_cols}. "
f"Available columns: {list(ipl_players_dag_df.columns)}"
)
if ipl_players_dag_df.empty:
raise ValueError('Ingested IPL player dataset is empty — nothing to process.')
churn_rate = ipl_players_dag_df['performance_churn'].mean()
print(f"Ingested {len(ipl_players_dag_df)} IPL player records.")
print(f"Performance churn rate in dataset: {churn_rate:.1%}")
print(ipl_players_dag_df[['match_id', 'player_name', 'batting_average', 'performance_churn']].head(5).to_string())
# Persist for downstream tasks
os.makedirs(os.path.dirname(STAGING_JSON), exist_ok=True)
ipl_players_dag_df.to_json(STAGING_JSON, orient='records', indent=2)
print(f"Staged data written to {STAGING_JSON}")
return STAGING_JSON
Step 3: Feature Engineering Task
The feature_engineering task transforms raw batting statistics into predictive features using the cricket_feature_pipeline function — a modular transformation function that encapsulates all feature logic. Three derived features are computed: batting_form_score normalizes batting average against the squad's maximum to give a relative form indicator, innings_density divides total runs by innings count to measure run-scoring consistency per innings, and boundary_efficiency measures what fraction of total runs came from boundaries as a proxy for aggressive intent. The feature matrix is serialized to Parquet for efficient reading in the training task.
Analogy🏏Cricket
🏏 Think of it like cricket: the feature engineering step is Virat Kohli's batting coach converting raw scorecards into the advanced statistics that actually predict churn — not just total runs, but runs per innings (innings_density), how often the player finds boundaries versus dots (boundary_efficiency), and how their average compares to the current squad baseline (batting_form_score). A player scoring 500 runs in 20 innings is very different from one scoring 500 in 8 innings. The cricket_feature_pipeline turns these human insights into numbers the model can use, just as a T20 analyst would extract strike-phase metrics from a raw scorecard before presenting to the coach. Keeping all of this inside one named cricket_feature_pipeline function is the coach writing his method into the team manual rather than doing arithmetic on a whiteboard: the same function can be unit-tested on a five-row practice scorecard, and — crucially — reused verbatim at prediction time, so the numbers the model sees in production are computed by exactly the same hand that computed its training data (no training-serving skew).
python
# dags/ipl_churn_pipeline.py — cricket_feature_pipeline helper + Task 2
import pandas as pd
import numpy as np
def cricket_feature_pipeline(ipl_players_dag_df: pd.DataFrame) -> pd.DataFrame:
"""
Modular feature engineering function for IPL player churn prediction.
Encapsulates all transformation logic; call from within the Airflow task
and also independently in unit tests.
"""
df = ipl_players_dag_df.copy()
# Feature 1: batting_form_score
# Normalised batting average — how strong is this player relative to squad peers?
max_batting_average = df['batting_average'].max()
df['batting_form_score'] = (df['batting_average'] / max_batting_average).clip(0, 1)
# Feature 2: innings_density
# Runs per innings — rewards consistent scorers over occasional big hitters
df['innings_density'] = (
df['total_runs'] / df['innings_count'].clip(lower=1)
).clip(0, 200)
# Feature 3: boundary_efficiency
# Fraction of runs from boundaries — measures aggressive batting style
# boundaries_hit are fours (4 runs each)
boundary_runs = df['boundaries_hit'] * 4
df['boundary_efficiency'] = (
boundary_runs / df['total_runs'].clip(lower=1)
).clip(0, 1)
# Feature 4: experience_index
# Normalised matches played — veterans vs newcomers behave differently under pressure
max_matches = df['matches_played'].max()
df['experience_index'] = (df['matches_played'] / max_matches).clip(0, 1)
return df
@task(task_id='feature_engineering')
def feature_engineering(staging_path: str) -> str:
"""
Load staged IPL player data, run cricket_feature_pipeline,
write feature matrix to Parquet. Returns features file path.
"""
ipl_players_dag_df = pd.read_json(staging_path, orient='records')
enriched_df = cricket_feature_pipeline(ipl_players_dag_df)
# Select final feature columns + label
feature_cols = [
'match_id',
'batting_average',
'innings_count',
'batting_form_score',
'innings_density',
'boundary_efficiency',
'experience_index',
'performance_churn' # label
]
feature_matrix = enriched_df[feature_cols].copy()
# Ensure no NaN or Inf values before training
feature_matrix.replace([np.inf, -np.inf], np.nan, inplace=True)
n_null_before = feature_matrix.isnull().sum().sum()
feature_matrix.fillna(0.0, inplace=True)
if n_null_before > 0:
print(f" Warning: {n_null_before} NaN values filled with 0.0")
os.makedirs(os.path.dirname(FEATURES_PARQUET), exist_ok=True)
feature_matrix.to_parquet(FEATURES_PARQUET, index=False)
print(f"Feature matrix: {feature_matrix.shape} — saved to {FEATURES_PARQUET}")
print("Feature means:")
for col in ['batting_form_score', 'innings_density', 'boundary_efficiency', 'experience_index']:
print(f" {col}: {feature_matrix[col].mean():.4f}")
return FEATURES_PARQUET
Step 4: Train and Register Model
The train_model task fits a GradientBoostingClassifier on the engineered feature matrix, evaluates it with AUC-ROC and F1 score, and logs all parameters and metrics to MLflow for experiment tracking. The subsequent register_model task queries the MLflow Model Registry to check whether there is a currently registered champion model. If the new model outperforms the champion on AUC-ROC by at least 0.5%, it is promoted to the Production stage in the registry. This two-task split cleanly separates training concerns from promotion concerns, matching the champion-challenger pattern discussed in the alerting lesson. The register task returns a summary dictionary that the notify task consumes.
Analogy🏏Cricket
🏏 Think of it like cricket: The train-and-register pair works like a selection trial with an incumbent. The trial itself (train_model) is fully documented — every drill setting and every measured result goes into the tour diary (MLflow logging of params and metrics) so any selector can later reconstruct exactly how the candidate was assessed. Then the selection meeting (register_model) applies one cold rule: the challenger only takes the incumbent's shirt if the measured AUC beats the version currently registered — no sentimentality, no 'but he trained hard'. If the challenger loses, the incumbent stays, the diary still records the attempt, and the pipeline carries on without a deployment. That conditional promotion is the single most protective habit in automated retraining: a bad data day produces a weak challenger, and the gate quietly refuses it instead of putting a worse model in front of the coach.
python
# dags/ipl_churn_pipeline.py — Tasks 3 and 4: train_model + register_model
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import roc_auc_score, f1_score
import mlflow
import mlflow.sklearn
import json
@task(task_id='train_model')
def train_model(features_path: str) -> str:
"""
Train GradientBoostingClassifier on IPL churn features.
Log parameters, metrics, and model artifact to MLflow.
Returns: path to training metrics JSON.
"""
feature_matrix = pd.read_parquet(features_path)
predictor_cols = [
'batting_average', 'innings_count',
'batting_form_score', 'innings_density',
'boundary_efficiency', 'experience_index'
]
X = feature_matrix[predictor_cols].values
y = feature_matrix['performance_churn'].values
X_train, X_val, y_train, y_val = train_test_split(
X, y, test_size=0.2, random_state=42,
stratify=y if len(np.unique(y)) > 1 else None
)
mlflow.set_tracking_uri(MLFLOW_TRACKING)
mlflow.set_experiment('ipl_player_churn_prediction')
with mlflow.start_run(run_name='ipl_churn_gbm') as run:
# Model hyperparameters
n_estimators = 150
learning_rate = 0.08
max_depth = 4
model = GradientBoostingClassifier(
n_estimators=n_estimators,
learning_rate=learning_rate,
max_depth=max_depth,
random_state=42
)
model.fit(X_train, y_train)
# Evaluation metrics
y_proba = model.predict_proba(X_val)[:, 1]
y_pred = (y_proba > 0.5).astype(int)
val_auc = float(roc_auc_score(y_val, y_proba))
val_f1 = float(f1_score(y_val, y_pred, zero_division=0))
# Log to MLflow
mlflow.log_params({
'n_estimators': n_estimators,
'learning_rate': learning_rate,
'max_depth': max_depth,
'n_features': len(predictor_cols),
})
mlflow.log_metrics({'val_auc': round(val_auc, 4), 'val_f1': round(val_f1, 4)})
mlflow.sklearn.log_model(
sk_model=model,
artifact_path='ipl_churn_model',
registered_model_name=MLFLOW_MODEL_NAME
)
run_id = run.info.run_id
print(f"Training complete — AUC: {val_auc:.4f}, F1: {val_f1:.4f}, run_id: {run_id}")
training_metrics = {
'run_id': run_id,
'val_auc': round(val_auc, 4),
'val_f1': round(val_f1, 4),
'n_train': int(len(X_train)),
'n_val': int(len(X_val)),
'n_estimators': n_estimators,
'learning_rate': learning_rate,
}
os.makedirs(os.path.dirname(METRICS_JSON), exist_ok=True)
with open(METRICS_JSON, 'w') as fh:
json.dump(training_metrics, fh, indent=2)
return METRICS_JSON
@task(task_id='register_model')
def register_model(metrics_path: str) -> dict:
"""
Compare challenger vs champion in MLflow Model Registry.
Promote challenger to Production if AUC improves by >= 0.005.
Returns registration summary dict for the notify_team task.
"""
with open(metrics_path) as fh:
challenger_metrics = json.load(fh)
challenger_auc = challenger_metrics['val_auc']
challenger_run_id = challenger_metrics['run_id']
mlflow.set_tracking_uri(MLFLOW_TRACKING)
client = mlflow.tracking.MlflowClient()
# Fetch current Production champion AUC from registry
champion_auc = None
try:
production_versions = client.get_latest_versions(
MLFLOW_MODEL_NAME, stages=['Production']
)
if production_versions:
champion_run = client.get_run(production_versions[0].run_id)
champion_auc = float(champion_run.data.metrics.get('val_auc', 0.0))
except Exception:
champion_auc = None # No champion exists yet — first run
promoted = False
reason = ''
min_improvement = 0.005
if champion_auc is None:
# No champion — promote challenger unconditionally
client.transition_model_version_stage(
name=MLFLOW_MODEL_NAME,
version=client.get_latest_versions(MLFLOW_MODEL_NAME)[0].version,
stage='Production'
)
promoted = True
reason = 'First registration — no existing champion'
elif challenger_auc > champion_auc + min_improvement:
client.transition_model_version_stage(
name=MLFLOW_MODEL_NAME,
version=client.get_latest_versions(MLFLOW_MODEL_NAME)[0].version,
stage='Production'
)
promoted = True
reason = f'Challenger AUC {challenger_auc:.4f} > champion {champion_auc:.4f} + {min_improvement}'
else:
reason = f'Challenger AUC {challenger_auc:.4f} did not beat champion {champion_auc} + {min_improvement}'
registration_summary = {
'model_name': MLFLOW_MODEL_NAME,
'run_id': challenger_run_id,
'challenger_auc': challenger_auc,
'champion_auc': champion_auc,
'promoted': promoted,
'reason': reason,
'val_f1': challenger_metrics['val_f1'],
}
print(f"Model registration: {'PROMOTED' if promoted else 'NOT PROMOTED'}")
print(f"Reason: {reason}")
return registration_summary
Testing and Verification
After completing the full DAG file with the notify_team task and task wiring, verify the pipeline end-to-end using the Airflow CLI. Test each task individually first with airflow tasks test before triggering a complete run. Check that MLflow's local tracking database is populated with experiment runs, that the model appears in the Model Registry under the Production stage after the first successful run, and that the notify_team task logs the expected summary message. Use the verification script below to automate the pipeline logic checks outside of Airflow's scheduler, enabling rapid iteration during development.
Analogy🏏Cricket
🏏 Think of it like cricket: verifying a monitoring system is testing the smoke alarm, and the only honest test is lighting a controlled fire. A stadium safety officer does not certify the fire system by admiring the control panel — she triggers a known, contained smoke source in bay 4 and confirms three things: the alarm for bay 4 fires (the planted drift is detected), the alarms for bays 1 through 3 stay silent (the stable features are not falsely flagged), and the control room receives the structured signal it needs to act (your extracted drift dictionary carries the right feature names and scores). Your verification script is precisely this drill: because you *engineered* the drift into specific columns in Step 1, you know exactly which alarms must ring, and a detector that fires everywhere — or nowhere — fails the drill regardless of how handsome the dashboard looks. Carry the habit forward: every production monitoring pipeline deserves a periodic planted-fault test, because the most dangerous monitoring system is not the one that is missing, but the one everybody trusts and nobody has ever seen catch a real fire.
python
# notify_team task + complete DAG wiring + verification script
import json
import os
from airflow.decorators import task
@task(task_id='notify_team')
def notify_team(registration_summary: dict) -> None:
"""
Send training results notification to the coaching staff channel.
In production: POST to Slack webhook URL stored in Airflow Variables.
In development: prints formatted summary to Airflow task logs.
"""
promoted = registration_summary['promoted']
challenger = registration_summary['challenger_auc']
champion = registration_summary['champion_auc']
val_f1 = registration_summary['val_f1']
model_name = registration_summary['model_name']
run_id = registration_summary['run_id'][:8] # abbreviated for readability
status_emoji = '[PROMOTED]' if promoted else '[NOT PROMOTED]'
notification_message = (
f"{status_emoji} IPL Player Churn Model — Daily Training Complete\n"
f" Model: {model_name}\n"
f" Run ID: {run_id}\n"
f" Challenger AUC: {challenger:.4f} | Champion AUC: {champion}\n"
f" Val F1: {val_f1:.4f}\n"
f" Promotion decision: {registration_summary['reason']}\n"
f" Dashboard: http://localhost:5000 (MLflow tracking)"
)
# Production: POST to Slack webhook
# import requests
# slack_webhook_url = Variable.get('slack_mlops_webhook_url')
# requests.post(slack_webhook_url, json={'text': notification_message})
print("=" * 60)
print(notification_message)
print("=" * 60)
# ── Complete DAG function body (replace 'pass' in the scaffold) ───────────────
# The five tasks are wired inside @dag via TaskFlow data passing:
#
# def ipl_players_dag():
# staging_path = ingest_ipl_data()
# features_path = feature_engineering(staging_path)
# metrics_path = train_model(features_path)
# registration_summary = register_model(metrics_path)
# notify_team(registration_summary)
# ── Standalone verification script (run outside Airflow during development) ──
def run_pipeline_standalone():
"""Execute the full pipeline logic outside Airflow for rapid testing."""
import pandas as pd
import numpy as np
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import roc_auc_score, f1_score
print("=== IPL Churn Pipeline — Standalone Verification ===")
# Step 1: Load data
ipl_players_dag_df = pd.read_csv('/tmp/ipl_pipeline/data/ipl_players_raw.csv')
print(f"[1/5] Ingested {len(ipl_players_dag_df)} records")
assert not ipl_players_dag_df.empty, 'Data is empty'
assert 'performance_churn' in ipl_players_dag_df.columns, 'Missing label column'
# Step 2: Feature engineering
ipl_players_dag_df['batting_form_score'] = (
ipl_players_dag_df['batting_average'] / ipl_players_dag_df['batting_average'].max()
)
ipl_players_dag_df['innings_density'] = (
ipl_players_dag_df['total_runs'] / ipl_players_dag_df['innings_count'].clip(1)
)
ipl_players_dag_df['boundary_efficiency'] = (
ipl_players_dag_df['boundaries_hit'] * 4 / ipl_players_dag_df['total_runs'].clip(1)
)
ipl_players_dag_df['experience_index'] = (
ipl_players_dag_df['matches_played'] / ipl_players_dag_df['matches_played'].max()
)
print(f"[2/5] Feature engineering complete — batting_form_score mean: "
f"{ipl_players_dag_df['batting_form_score'].mean():.4f}")
# Step 3: Train model
predictor_cols = ['batting_average', 'innings_count', 'batting_form_score',
'innings_density', 'boundary_efficiency', 'experience_index']
X = ipl_players_dag_df[predictor_cols].fillna(0).values
y = ipl_players_dag_df['performance_churn'].values
X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.2, random_state=42)
model = GradientBoostingClassifier(n_estimators=150, learning_rate=0.08, random_state=42)
model.fit(X_train, y_train)
val_auc = roc_auc_score(y_val, model.predict_proba(X_val)[:, 1])
val_f1 = f1_score(y_val, model.predict(X_val), zero_division=0)
print(f"[3/5] Model trained — AUC: {val_auc:.4f}, F1: {val_f1:.4f}")
assert val_auc > 0.5, f'AUC {val_auc:.4f} is not better than random'
# Step 4: Simulate registration (no MLflow in standalone)
simulated_registration = {
'model_name': MLFLOW_MODEL_NAME, 'run_id': 'standalone_test',
'challenger_auc': round(val_auc, 4), 'champion_auc': 0.72,
'promoted': val_auc > 0.72 + 0.005,
'reason': 'standalone verification run', 'val_f1': round(val_f1, 4)
}
print(f"[4/5] Registration: {'PROMOTED' if simulated_registration['promoted'] else 'NOT PROMOTED'}")
# Step 5: Notification
print(f"[5/5] Team notification: AUC={val_auc:.4f}, F1={val_f1:.4f}")
print("=== All 5 pipeline steps verified successfully ===")
print("Safe to deploy to Airflow: copy ipl_churn_pipeline.py to your dags/ directory")
print("Then run: airflow dags trigger ipl_player_churn_pipeline")
# Run standalone verification
run_pipeline_standalone()
Warning: When using MLflow's Model Registry with a local SQLite backend (sqlite:////tmp/...), concurrent Airflow tasks writing to the same database will cause database lock errors. In production, always point MLFLOW_TRACKING_URI to a PostgreSQL or MySQL backend that handles concurrent writes correctly. For local development, this exercise uses sequential tasks so SQLite is safe, but never use SQLite for multi-worker Airflow setups running parallel training jobs.
Pro Tip
Use Airflow Variables (Admin > Variables in the UI) to store all configurable parameters outside the DAG code — RAW_DATA_PATH, MLFLOW_TRACKING_URI, Slack webhook URL, and the minimum AUC improvement threshold. Retrieve them inside tasks with Variable.get('ipl_raw_data_path', default_var='...'). This makes promoting the DAG across environments (dev > staging > prod) painless: each environment's Airflow instance holds its own Variable values pointing to the right data sources and thresholds without any code changes.
- You built a five-task Airflow DAG using the TaskFlow API covering the full ML lifecycle: ingest_ipl_data, feature_engineering, train_model, register_model, and notify_team, wired with the >> operator.
- The cricket_feature_pipeline function encapsulates all feature engineering logic as a testable, reusable unit — call it from the Airflow task and independently in unit tests without Airflow overhead.
- GradientBoostingClassifier logged to MLflow with mlflow.sklearn.log_model enables experiment tracking, parameter logging, and model artifact versioning in a single context manager block.
- The champion-challenger register_model task queries the MLflow Model Registry for the current Production model's AUC and only promotes the challenger if it improves by at least 0.5%, preventing regressions from bad retraining runs.
- The notify_team task decouples stakeholder communication from training logic — in production, swap the print statement for a Slack webhook POST using a URL stored in an Airflow Variable, requiring zero code changes to the training tasks.
- Setting catchup=False on a daily-scheduled DAG prevents Airflow from spawning hundreds of backfill runs when the DAG is first enabled — critical for ML pipelines where each run consumes significant compute resources.