Add Airflow DAG for training pipeline orchestration

airflow/dags/ml_training_pipeline_dag.py

@@ -0,0 +1,143 @@
+from datetime import datetime, timedelta
+from airflow import DAG
+from airflow.operators.python import PythonOperator
+import sys
+from pathlib import Path
+
+# Add project root to path
+project_root = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(project_root))
+
+from mlpipeline.logging.logger import get_logger
+from mlpipeline.stages.data_ingestion_pipeline import DataIngestionPipeline
+from mlpipeline.stages.data_validation_pipeline import DataValidationPipeline
+from mlpipeline.stages.data_transformation_pipeline import DataTransformationPipeline
+from mlpipeline.stages.feature_engineering_pipeline import FeatureEngineeringPipeline
+from mlpipeline.stages.model_trainer_pipeline import ModelTrainerPipeline
+from mlpipeline.stages.model_evaluation_pipeline import ModelEvaluationPipeline
+from mlpipeline.stages.model_pusher_pipeline import ModelPusherPipeline
+
+logger = get_logger(__name__)
+
+default_args = {
+    'owner': 'automl-mlops',
+    'depends_on_past': False,
+    'start_date': datetime(2026, 1, 1),
+    'email_on_failure': False,
+    'email_on_retry': False,
+    'retries': 1,
+    'retry_delay': timedelta(minutes=5)
+}
+
+dag = DAG(
+    'automl_training_pipeline',
+    default_args=default_args,
+    description='AutoML Training Pipeline with MLflow Tracking',
+    schedule_interval='@daily',
+    catchup=False,
+    tags=['automl', 'mlops', 'mlflow']
+)
+
+
+def run_data_ingestion():
+    logger.info("Stage 1: Starting Data Ingestion")
+    pipeline = DataIngestionPipeline()
+    pipeline.run()
+    logger.info("Stage 1: Data Ingestion completed")
+
+
+def run_data_validation():
+    logger.info("Stage 2: Starting Data Validation")
+    pipeline = DataValidationPipeline()
+    pipeline.run()
+    logger.info("Stage 2: Data Validation completed")
+
+
+def run_data_transformation():
+    logger.info("Stage 3: Starting Data Transformation")
+    pipeline = DataTransformationPipeline()
+    pipeline.run()
+    logger.info("Stage 3: Data Transformation completed")
+
+
+def run_feature_engineering():
+    logger.info("Stage 4: Starting Feature Engineering")
+    pipeline = FeatureEngineeringPipeline()
+    pipeline.run()
+    logger.info("Stage 4: Feature Engineering completed")
+
+
+def run_model_trainer():
+    logger.info("Stage 5: Starting Model Training")
+    pipeline = ModelTrainerPipeline()
+    pipeline.run()
+    logger.info("Stage 5: Model Training completed")
+
+
+def run_model_evaluation():
+    logger.info("Stage 6: Starting Model Evaluation")
+    pipeline = ModelEvaluationPipeline()
+    pipeline.run()
+    logger.info("Stage 6: Model Evaluation completed")
+
+
+def run_model_pusher():
+    logger.info("Stage 7: Starting Model Pusher")
+    pipeline = ModelPusherPipeline()
+    pipeline.run()
+    logger.info("Stage 7: Model Pusher completed")
+
+
+# Define tasks
+data_ingestion_task = PythonOperator(
+    task_id='data_ingestion',
+    python_callable=run_data_ingestion,
+    dag=dag
+)
+
+data_validation_task = PythonOperator(
+    task_id='data_validation',
+    python_callable=run_data_validation,
+    dag=dag
+)
+
+data_transformation_task = PythonOperator(
+    task_id='data_transformation',
+    python_callable=run_data_transformation,
+    dag=dag
+)
+
+feature_engineering_task = PythonOperator(
+    task_id='feature_engineering',
+    python_callable=run_feature_engineering,
+    dag=dag
+)
+
+model_trainer_task = PythonOperator(
+    task_id='model_trainer',
+    python_callable=run_model_trainer,
+    dag=dag
+)
+
+model_evaluation_task = PythonOperator(
+    task_id='model_evaluation',
+    python_callable=run_model_evaluation,
+    dag=dag
+)
+
+model_pusher_task = PythonOperator(
+    task_id='model_pusher',
+    python_callable=run_model_pusher,
+    dag=dag
+)
+
+# Define pipeline dependencies
+(
+    data_ingestion_task 
+    >> data_validation_task 
+    >> data_transformation_task 
+    >> feature_engineering_task 
+    >> model_trainer_task 
+    >> model_evaluation_task 
+    >> model_pusher_task
+)

requirements.txt

@@ -15,6 +15,7 @@ pycaret
 
 mlflow
 dvc
+apache-airflow
 
 evidently