Airflow Implemented

airflow/airflow.cfg

@@ -0,0 +1,18 @@
+# Minimal Airflow configuration for local development
+
+[core]
+dags_folder = d:\AutoML MLOps PipeLine\airflow\dags
+base_log_folder = d:\AutoML MLOps PipeLine\airflow\logs
+executor = LocalExecutor
+load_examples = False
+parallelism = 2
+
+[webserver]
+web_server_port = 8080
+reload_on_plugin_change = True
+
+[scheduler]
+catchup_by_default = False
+
+[api]
+auth_backend = airflow.api.auth.backend.basic_auth

airflow/dags/ml_training_pipeline_dag.py

@@ -5,26 +5,19 @@ 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))
+sys.path.insert(0, str(Path(__file__).parent.parent.parent))
 
+from mlpipeline.pipeline.training_pipeline import TrainingPipeline
 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__)
 
+# DAG configuration
 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)
 }
@@ -38,106 +31,16 @@ dag = DAG(
     tags=['automl', 'mlops', 'mlflow']
 )
 
+# Single function to run entire pipeline
+def run_training_pipeline():
+    logger.info("Starting AutoML Training Pipeline")
+    pipeline = TrainingPipeline()
+    pipeline.run_pipeline()
+    logger.info("Training Pipeline completed")
 
-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,
+# Single task for entire pipeline
+training_task = PythonOperator(
+    task_id='automl_training_pipeline',
+    python_callable=run_training_pipeline,
     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
-)