Machines don\u2019t break down out of nowhere\u2014there are always signs. The problem? Humans aren\u2019t always great at noticing them. That\u2019s where Machine Predictive Maintenance comes in! This guide will take you through the exciting world of Machine Predictive Maintenance, using AWS and MLOps <\/a>to ensure your equipment stays predictably reliable.<\/p>\n This article was published as a part of the\u00a0<\/strong><\/em>Data Science Blogathon.<\/strong><\/em><\/a><\/p>\n Unexpected equipment failures in industrial settings cause downtime and financial losses. In our project, we are using MLOps best practices and machine learning to detect issues early, enabling timely repairs and reducing disruptions.<\/p>\n Before diving into implementation, let\u2019s take a closer look at the project architecture.<\/p>\n Below we will first look in the prerequisites required:<\/p>\n Clone the repository<\/strong>:<\/p>\n Create and activate the virtual environment<\/strong>:<\/p>\n Install Required Dependencies<\/strong>:<\/p>\n Set Up Environment Variables:<\/strong><\/p>\n The Project Structure outlines the key components and organization of the project, ensuring clarity and maintainability. It helps in understanding how different modules interact and how the overall system is designed. A well-defined structure simplifies development, debugging, and scalability.<\/p>\n In this project, we use a machine predictive maintenance CSV file, converting it into JSON records, and inserting it into a MongoDB collection.\u00a0<\/p>\n Dataset Link: https:\/\/www.kaggle.com\/datasets\/shivamb\/machine-predictive-maintenance-classification<\/a><\/p>\n Here is the code snippet to convert CSV -> JSON records -> MongoDB<\/p>\n Here is the code snippet to fetch data from MongoDB, split the data into train and test CSV files, and store them as a Data Ingestion artifact.<\/p>\n In this step, we\u2019ll check if the ingested data meets the expected format by ensuring all required columns are present using a predefined schema and comparing the training and testing data for any differences. Then we save the clean data and create a drift report, so only quality data is used for the next step, which is transforming the data.\u00a0<\/p>\n Data Validation code snippet:<\/p>\n Below is the drift report generated by the Data Validation.<\/p>\n Here, we will clean and transform the validated data by converting temperature features from Kelvin to Celsius, dropping unnecessary columns, and applying a transformation pipeline that uses ordinal encoding and Min-Max scaling, and handling data imbalance using SMOTEENN.<\/p>\n The transformed training and test datasets are saved as .npy files<\/b> along with a serialized preprocessing object i.e. Minmax scaler(preprocessing.pkl<\/b>) all encapsulated as an artifact for further model training.<\/p>\n Now we will train different classification models on the transformed training data like Decision Tree,\u00a0Random Forest,\u00a0Gradient Boosting,\u00a0Logistic Regression, and AdaBoost models and evaluate their performance using evaluation metrics like f1-score, precision, and recall, and then log these details with MLflow<\/a>.<\/p>\n After selecting the best model, it saves both the model and transformation object locally to artifact and evaluation metrics to use it for deployment.<\/p>\n Now, let\u2019s create a training_pipeline.py<\/b> file where we sequentially integrate all the steps of data ingestion, validation, transformation, and model training into a complete pipeline.<\/p>\n You can visually see how we have built the training_pipeline<\/a>.<\/p>\n Now that we\u2019ve completed creating the pipeline, run the training_pipeline.py file to view the artifacts generated in the previous steps.<\/p>\n Run the following command to view the MLflow dashboard.<\/p>\n As you can see, we have successfully logged model metrics like recall, precision, and F1-score in MLflow.<\/p>\n After running the training_pipeline.py file, we generated artifacts and stored them locally. Now, we will store these artifacts in an AWS S3 bucket to enable cloud-based storage and accessibility.<\/p>\n Using a Docker image, we will push it to AWS ECR via GitHub Actions and then deploy it to production using AWS EC2. We will discuss this process in more detail in the upcoming sections.<\/p>\n Follow these steps to create an AWS S3 bucket:<\/p>\n Your IAM user has been successfully connected to your project.<\/p>\n Your bucket is now created, and you can start uploading artifacts. Now, add the following code to your training_pipeline.py file and run it again to see the artifacts in your AWS S3 bucket.<\/p>\n Follow these steps to create an ECR repository.<\/p>\n Copy the URI from your ECR repository, which should look something like \u201c788614365622.dkr.ecr.ap-southeast-2.amazonaws.com<\/b>\u201c, and save it somewhere. We will need it later to paste into GitHub Secrets.<\/p>\n Dockerizing our project ensures it runs smoothly in any environment without dependency issues. It\u2019s a must-have tool for packaging and sharing applications effortlessly.<\/p>\n Keeping the Docker image as small as possible is important for efficiency. We can minimize its size by using techniques like multi-staging and choosing a lightweight Python base image.<\/p>\n To securely store sensitive information like AWS credentials and repository URIs, we need to set up GitHub Action Secrets in our repository. Follow these steps:<\/p>\n Navigate to your repository on GitHub.<\/p>\n At the top of your repository page, locate and click on the \u201cSettings\u201d tab.<\/p>\n In the left sidebar, scroll down to Secrets and Variables \u2192 Select Actions.<\/p>\n Click the New repository secret button.<\/p>\n Now let\u2019s understand how to create an instance with AWS EC2. Follow these steps.<\/p>\n After creating the instance named machinepredictive, select its Instance ID and click Connect. Then, under EC2 Instance Connect, click Connect again. This will open an AWS CLI interface where you can run the necessary commands.<\/p>\n Now, enter these commands in the CLI one by one to set up Docker on your EC2 instance:<\/p>\n Keep your CLI session open. Next, navigate to your GitHub repository, click on Settings, then go to the Actions section, and under Runners, click New self-hosted runner. Choose the Linux runner image. Now, run the following commands one by one in your CLI to download and configure your self-hosted runner:<\/p>\n This will download, configure, and start your self-hosted GitHub Actions runner on your EC2 instance.<\/p>\n After setting up the self-hosted GitHub Actions runner, the CLI will prompt you to enter a name for the runner. Type self-hosted and press Enter. <\/p>\n You can check out the .github\/workflows\/main.yml<\/b> code<\/a>\u00a0file.<\/p>\n Now, let\u2019s dive into what each section of this main.yml file does.<\/p>\n Once the Continuous Deployment Job completes successfully, you\u2019ll see an output like this in the AWS CLI.<\/p>\n Open your AWS EC2 instance by clicking its Instance ID. Verify that the instance state is Running and locate the Public IPv4 DNS. Click on the \u201cOpen address\u201d option to automatically launch your FastAPI application in your browser. Now, let\u2019s dive into FastAPI.<\/p>\n You can check out the app.py<\/b> file<\/a>.\u00a0<\/p>\n In this FastAPI application, I\u2019ve created two main routes, one for training the model and another for generating predictions. Let\u2019s explore each route in detail.\u00a0\u00a0<\/p>\n The \/train endpoint starts the model training process using a predefined dataset, making it easy to update and improve the model with new data.\u00a0It is especially useful for retraining the model to improve accuracy or incorporate new data, ensuring the model remains up-to-date and performs optimally.<\/p>\n The \/predict endpoint accepts a CSV file via POST. This endpoint handles incoming data, leverages the trained model to generate predictions, and then returns the outcomes formatted as JSON.\u00a0This route is perfect for applying the model to new datasets, making it efficient for large-scale prediction tasks.<\/p>\n In the \/predict route, we\u2019ve included a sample test.csv file; you can download it here<\/a>.<\/p>\n Together, we built a full production-ready Predictive Maintenance MLOps project\u2014from gathering and preprocessing data to training, evaluating, and deploying our model using Docker, AWS, and FastAPI. This project shows how MLOps can bridge the gap between development and production, making it easier to build robust, scalable solutions.<\/p>\n Remember, this guide is all about learning and applying these techniques to your data science projects. Don\u2019t hesitate to experiment, innovate, and explore new enhancements as you progress. Thank you for sticking with me until the end\u2014keep learning, keep doing, and keep growing!<\/p>\nLearning Objectives<\/h3>\n
\n
Problem: Unplanned Downtime & Maintenance Costs<\/h2>\n
![\"machine_Predictive_Maintenance\"](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2025\/02\/machine_Predictive_Maintenance.webp\")
<\/figure>\nNecessary Prerequisites<\/h2>\n
git clone \"https:\/\/github.com\/karthikponna\/Predictive_Maintenance_MLOps.git\"\ncd Predictive_Maintenance_MLOps<\/code><\/pre>\n# For macOS and Linux:\npython3 -m venv venv\nsource venv\/bin\/activate\n\n# For Windows:\npython -m venv venv\n.\\venv\\Scripts\\activate<\/code><\/pre>\npip install -r requirements.txt<\/code><\/pre>\n# Create a `.env` file and add your MongoDB connection string:\nMONGO_URI=your_mongodb_connection_string<\/code><\/pre>\nProject Structure<\/h2>\n
project_root\/\n\u2502\n\u251c\u2500\u2500 .github\/\n\u2502 \u2514\u2500\u2500 workflows\/\n\u2502 \u2514\u2500\u2500 main.yml\n\u2502\n\u251c\u2500\u2500 data_schema\/\n\u2502 \u2514\u2500\u2500 schema.yaml\n\u2502\n\u251c\u2500\u2500 final_model\/\n\u2502 \u251c\u2500\u2500 model.pkl\n\u2502 \u2514\u2500\u2500 preprocessor.pkl\n\u2502\n\u251c\u2500\u2500 Machine_Predictive_Data\/\n\u2502 \u2514\u2500\u2500 predictive_maintenance.csv\n\u2502\n\u251c\u2500\u2500 machine_predictive_maintenance\/\n\u2502 \u251c\u2500\u2500 cloud\/\n\u2502 \u251c\u2500\u2500 components\/\n\u2502 \u251c\u2500\u2500 constant\/\n\u2502 \u251c\u2500\u2500 entity\/\n\u2502 \u251c\u2500\u2500 exception\/\n\u2502 \u251c\u2500\u2500 logging\/\n\u2502 \u251c\u2500\u2500 pipeline\/\n\u2502 \u251c\u2500\u2500 utils\/\n\u2502 \u2514\u2500\u2500 __init__.py\n\u2502\n\u251c\u2500\u2500 my_venv\/\n\u2502\n\u251c\u2500\u2500 notebooks\/\n\u2502 \u251c\u2500\u2500 EDA.ipynb\n\u2502 \u251c\u2500\u2500 prediction_output\/\n\u2502\n\u251c\u2500\u2500 templates\/\n\u2502 \u2514\u2500\u2500 table.html\n\u2502\n\u251c\u2500\u2500 valid_data\/\n| \u2514\u2500\u2500test.csv\n\u2502\n\u251c\u2500\u2500 .env\n\u251c\u2500\u2500 .gitignore\n\u251c\u2500\u2500 app.py\n\u251c\u2500\u2500 Dockerfile\n\u251c\u2500\u2500 main.py\n\u251c\u2500\u2500 push_data.py\n\u251c\u2500\u2500 README.md\n\u251c\u2500\u2500 requirements.txt\n\u251c\u2500\u2500 setup.py\n\u251c\u2500\u2500 test_mongodb.py<\/code><\/pre>\nData Ingestion<\/h2>\n
class PredictiveDataExtract():\n\n def __init__(self):\n \"\"\"\n Initializes the PredictiveDataExtract class.\n \"\"\"\n try:\n pass\n except Exception as e:\n raise MachinePredictiveMaintenanceException(e, sys)\n \n def csv_to_json_convertor(self, file_path):\n try:\n data = pd.read_csv(file_path)\n data.reset_index(drop=True, inplace=True)\n records = list(json.loads(data.T.to_json()).values())\n return records\n except Exception as e:\n raise MachinePredictiveMaintenanceException(e, sys)\n \n def insert_data_mongodb(self, records, database, collection):\n try:\n self.records = records\n self.database = database\n self.collection = collection\n\n self.mongo_client = pymongo.MongoClient(MONGO_DB_URL)\n self.database = self.mongo_client[self.database]\n \n self.collection = self.database[self.collection]\n self.collection.insert_many(self.records)\n return(len(self.records))\n except Exception as e:\n raise MachinePredictiveMaintenanceException(e, sys)\n \nif __name__==\"__main__\":\n FILE_PATH=\"Machine_Predictive_Data\/predictive_maintenance.csv\"\n DATABASE=\"Predictive_Maintenance_MLOps\"\n collection = \"Machine_Predictive_Data\"\n\n predictive_data_obj=PredictiveDataExtract()\n records = predictive_data_obj.csv_to_json_convertor(FILE_PATH)\n no_of_records = predictive_data_obj.insert_data_mongodb(records, DATABASE, collection)\n print(no_of_records)<\/code><\/pre>\nclass DataIngestion:\n def __init__(self, data_ingestion_config:DataIngestionConfig):\n\n \"\"\"\n Initializes the DataIngestion class with the provided configuration.\n\n Parameters:\n data_ingestion_config: DataIngestionConfig\n Configuration object containing details for data ingestion.\n \"\"\"\n try:\n self.data_ingestion_config=data_ingestion_config\n except Exception as e:\n raise MachinePredictiveMaintenanceException(e, sys)\n \n def export_collection_as_dataframe(self):\n try:\n database_name= self.data_ingestion_config.database_name\n collection_name= self.data_ingestion_config.collection_name\n self.mongo_client = pymongo.MongoClient(MONGO_DB_URL)\n collection = self.mongo_client[database_name][collection_name]\n df = pd.DataFrame(list(collection.find()))\n if \"_id\" in df.columns.to_list():\n df = df.drop(columns=[\"_id\"], axis=1)\n df.replace({\"na\":np.nan},inplace=True)\n return df\n except Exception as e:\n raise MachinePredictiveMaintenanceException(e, sys)\n\n def export_data_into_feature_store(self,dataframe: pd.DataFrame):\n try:\n feature_store_file_path=self.data_ingestion_config.feature_store_file_path\n #creating folder\n dir_path = os.path.dirname(feature_store_file_path)\n os.makedirs(dir_path,exist_ok=True)\n dataframe.to_csv(feature_store_file_path,index=False,header=True)\n return dataframe\n except Exception as e:\n raise MachinePredictiveMaintenanceException(e,sys)\n\n def split_data_as_train_test(self, dataframe:pd.DataFrame): \n try:\n train_set, test_set = train_test_split(\n dataframe, test_size=self.data_ingestion_config.train_test_split_ratio\n )\n logging.info(\"Performed train test split on the dataframe\")\n logging.info(\n \"Exited split_data_as_train_test method of Data_Ingestion class\"\n )\n dir_path = os.path.dirname(self.data_ingestion_config.training_file_path)\n os.makedirs(dir_path, exist_ok=True)\n logging.info(f\"Exporting train and test file path.\")\n train_set.to_csv(\n self.data_ingestion_config.training_file_path, index=False, header=True\n )\n test_set.to_csv(\n self.data_ingestion_config.testing_file_path, index=False, header=True\n )\n logging.info(f\"Exported train and test file path.\" )\n except Exception as e:\n raise MachinePredictiveMaintenanceException(e, sys)\n\n def initiate_data_ingestion(self):\n try:\n dataframe = self.export_collection_as_dataframe()\n dataframe = self.export_data_into_feature_store(dataframe)\n self.split_data_as_train_test(dataframe)\n dataingestionartifact= DataIngestionArtifact(trained_file_path=self.data_ingestion_config.training_file_path,\n test_file_path=self.data_ingestion_config.testing_file_path)\n return dataingestionartifact\n except Exception as e:\n raise MachinePredictiveMaintenanceException(e,sys)<\/code><\/pre>\nData Validation<\/h2>\n
class DataValidation:\n\n def __init__(self, data_ingestion_artifact:DataIngestionArtifact,\n data_validation_config: DataValidationConfig):\n try:\n self.data_ingestion_artifact= data_ingestion_artifact\n self.data_validation_config= data_validation_config\n self._schema_config = read_yaml_file(SCHEMA_FILE_PATH)\n except Exception as e:\n raise MachinePredictiveMaintenanceException(e, sys)\n \n @staticmethod\n def read_data(file_path) -> pd.DataFrame:\n try:\n return pd.read_csv(file_path)\n except Exception as e:\n raise MachinePredictiveMaintenanceException(e, sys) \n\n \n def validate_number_of_columns(self, dataframe:pd.DataFrame)-> bool:\n try:\n number_of_columns = len(self._schema_config)\n logging.info(f\"Required number of columns:{number_of_columns}\")\n logging.info(f\"Data frame has columns:{len(dataframe.columns)}\")\n\n if len(dataframe.columns) == number_of_columns: \n return True\n return False\n except Exception as e:\n raise MachinePredictiveMaintenanceException (e, sys)\n \n\n def is_columns_exist(self, df:pd.DataFrame) -> bool:\n try:\n dataframe_columns = df.columns\n missing_numerical_columns = []\n missing_categorical_columns = []\n\n for column in self._schema_config[\"numerical_columns\"]:\n if column not in dataframe_columns:\n missing_numerical_columns.append(column)\n\n if len(missing_numerical_columns) > 0:\n logging.info(f\"Missing numerical column: {missing_numerical_columns}\")\n\n for column in self._schema_config[\"categorical_columns\"]:\n if column not in dataframe_columns:\n missing_categorical_columns.append(column)\n\n if len(missing_categorical_columns) > 0:\n logging.info(f\"Missing categorical column: {missing_categorical_columns}\")\n return False if len(missing_categorical_columns)>0 or len(missing_numerical_columns)>0 else True\n \n except Exception as e:\n raise MachinePredictiveMaintenanceException(e, sys)\n\n \n def detect_dataset_drift(self, base_df, current_df, threshold = 0.05) -> bool :\n try:\n status = True\n report = {}\n for column in base_df.columns:\n d1 = base_df[column]\n d2 = current_df[column]\n\n is_same_dist = ks_2samp(d1, d2)\n if threshold DataValidationArtifact:\n try:\n validation_error_msg = \"\"\n logging.info(\"Starting data validation\")\n\n train_file_path = self.data_ingestion_artifact.trained_file_path\n test_file_path = self.data_ingestion_artifact.test_file_path\n\n # read the data from the train and test \n train_dataframe = DataValidation.read_data(train_file_path)\n test_dataframe = DataValidation.read_data(test_file_path)\n\n # validate number of columns\n status = self.validate_number_of_columns(dataframe=train_dataframe)\n logging.info(f\"All required columns present in training dataframe: {status}\")\n\n if not status:\n validation_error_msg += f\"Train dataframe does not contain all columns.\\n\"\n\n status = self.validate_number_of_columns(dataframe=test_dataframe)\n\n if not status:\n validation_error_msg += f\"Test dataframe does not contain all columns.\\n\"\n\n status = self.is_columns_exist(df=train_dataframe)\n\n if not status:\n validation_error_msg += f\"Columns are missing in training dataframe.\"\n\n status = self.is_columns_exist(df=test_dataframe)\n\n if not status:\n validation_error_msg += f\"columns are missing in test dataframe.\"\n ## lets check datadrift\n status=self.detect_dataset_drift(base_df=train_dataframe,current_df=test_dataframe)\n dir_path=os.path.dirname(self.data_validation_config.valid_train_file_path)\n os.makedirs(dir_path,exist_ok=True)\n\n train_dataframe.to_csv(\n self.data_validation_config.valid_train_file_path, index=False, header=True\n\n )\n test_dataframe.to_csv(\n self.data_validation_config.valid_test_file_path, index=False, header=True\n )\n\n data_validation_artifact = DataValidationArtifact(\n validation_status=status,\n valid_train_file_path=self.data_validation_config.valid_train_file_path,\n valid_test_file_path=self.data_validation_config.valid_test_file_path,\n invalid_train_file_path=None,\n invalid_test_file_path=None,\n drift_report_file_path=self.data_validation_config.drift_report_file_path,\n )\n return data_validation_artifact\n except Exception as e:\n raise MachinePredictiveMaintenanceException(e,sys)<\/code><\/pre>\nAir temperature [K]:\n drift_status: true\n p_value: 0.016622943467175914\nFailure Type:\n drift_status: false\n p_value: 1.0\nProcess temperature [K]:\n drift_status: false\n p_value: 0.052940072765804994\nProduct ID:\n drift_status: false\n p_value: 0.09120557172716418\nRotational speed [rpm]:\n drift_status: false\n p_value: 0.2673520066245566\nTarget:\n drift_status: false\n p_value: 0.999999998717466\nTool wear [min]:\n drift_status: false\n p_value: 0.13090856779628832\nTorque [Nm]:\n drift_status: false\n p_value: 0.5001773464540389\nType:\n drift_status: false\n p_value: 1.0\nUDI:\n drift_status: true\n p_value: 0.022542489133976953<\/code><\/pre>\nData Transformation<\/h2>\n
class DataTransformation:\n\n def __init__(self,data_validation_artifact: DataValidationArtifact,\n data_transformation_config: DataTransformationConfig):\n try:\n self.data_validation_artifact: DataValidationArtifact = data_validation_artifact\n self.data_transformation_config: DataTransformationConfig = data_transformation_config\n self._schema_config = read_yaml_file(file_path=SCHEMA_FILE_PATH)\n except Exception as e:\n raise MachinePredictiveMaintenanceException(e,sys)\n \n @staticmethod\n def read_data(file_path) -> pd.DataFrame:\n try:\n return pd.read_csv(file_path)\n except Exception as e:\n raise MachinePredictiveMaintenanceException(e, sys)\n \n def get_data_transformer_object(self):\n try:\n logging.info(\"Got numerical cols from schema config\")\n \n scaler = MinMaxScaler()\n # Fetching categories for OrdinalEncoder from schema config\n ordinal_categories = self._schema_config.get('ordinal_categories', [])\n ordinal_encoder = OrdinalEncoder(categories=ordinal_categories)\n\n logging.info(\"Initialized MinMaxScaler, OrdinalEncoder with categories\")\n\n ordinal_columns = self._schema_config['ordinal_columns']\n scaling_features = self._schema_config['scaling_features']\n preprocessor = ColumnTransformer(\n [\n (\"Ordinal_Encoder\", ordinal_encoder, ordinal_columns),\n (\"MinMaxScaling\", scaler, scaling_features)\n ]\n )\n logging.info(\"Created preprocessor object from ColumnTransformer\")\n return preprocessor\n except Exception as e:\n raise MachinePredictiveMaintenanceException(e, sys)\n \n def initiate_data_transformation(self) -> DataTransformationArtifact:\n try:\n \n logging.info(\"Starting data transformation\")\n preprocessor = self.get_data_transformer_object()\n train_df = DataTransformation.read_data(self.data_validation_artifact.valid_train_file_path)\n test_df = DataTransformation.read_data(self.data_validation_artifact.valid_test_file_path)\n input_feature_train_df = train_df.drop(columns=[TARGET_COLUMN], axis=1)\n target_feature_train_df = train_df[TARGET_COLUMN]\n\n input_feature_train_df['Air temperature [c]'] = input_feature_train_df['Air temperature [K]'] - 273.15\n input_feature_train_df['Process temperature [c]'] = input_feature_train_df['Process temperature [K]'] - 273.15\n drop_cols = self._schema_config['drop_columns']\n\n input_feature_train_df = drop_columns(df=input_feature_train_df, cols = drop_cols)\n logging.info(\"Completed dropping the columns for Training dataset\")\n input_feature_test_df = test_df.drop(columns=[TARGET_COLUMN], axis=1)\n target_feature_test_df = test_df[TARGET_COLUMN]\n\n input_feature_test_df['Air temperature [c]'] = input_feature_test_df['Air temperature [K]'] - 273.15\n input_feature_test_df['Process temperature [c]'] = input_feature_test_df['Process temperature [K]'] - 273.15\n drop_cols = self._schema_config['drop_columns']\n input_feature_test_df = drop_columns(df=input_feature_test_df, cols = drop_cols)\n\n logging.info(\"Completed dropping the columns for Testing dataset\")\n input_feature_train_arr = preprocessor.fit_transform(input_feature_train_df)\n input_feature_test_arr = preprocessor.transform(input_feature_test_df)\n \n smt = SMOTEENN(sampling_strategy=\"minority\")\n input_feature_train_final, target_feature_train_final = smt.fit_resample(\n input_feature_train_arr, target_feature_train_df\n )\n logging.info(\"Applied SMOTEENN on training dataset\")\n\n input_feature_test_final, target_feature_test_final = smt.fit_resample(\n input_feature_test_arr, target_feature_test_df\n )\n train_arr = np.c_[\n input_feature_train_final, np.array(target_feature_train_final)\n ]\n test_arr = np.c_[\n input_feature_test_final, np.array(target_feature_test_final)\n ]\n save_numpy_array_data(self.data_transformation_config.transformed_train_file_path, array=train_arr, )\n save_numpy_array_data(self.data_transformation_config.transformed_test_file_path,array=test_arr,)\n save_object( self.data_transformation_config.transformed_object_file_path, preprocessor,)\n\n save_object( \"final_model\/preprocessor.pkl\", preprocessor,)\n\n data_transformation_artifact=DataTransformationArtifact(\n transformed_object_file_path=self.data_transformation_config.transformed_object_file_path,\n transformed_train_file_path=self.data_transformation_config.transformed_train_file_path,\n transformed_test_file_path=self.data_transformation_config.transformed_test_file_path\n )\n return data_transformation_artifact\n except Exception as e:\n raise MachinePredictiveMaintenanceException(e, sys)<\/code><\/pre>\nTraining and Evaluation<\/h2>\n
class ModelTrainer:\n def __init__(self, data_transformation_artifact: DataTransformationArtifact, model_trainer_config: ModelTrainerConfig):\n try:\n self.data_transformation_artifact = data_transformation_artifact\n self.model_trainer_config = model_trainer_config\n\n except Exception as e:\n raise MachinePredictiveMaintenanceException(e, sys)\n \n def track_mlflow(self, best_model, classification_metric, input_example):\n \"\"\"\n Log model and metrics to MLflow.\n\n Args:\n best_model: The trained model object.\n classification_metric: The classification metrics (f1, precision, recall).\n input_example: An example input data sample for the model.\n \"\"\"\n try:\n with mlflow.start_run():\n f1_score=classification_metric.f1_score\n precision_score=classification_metric.precision_score\n recall_score=classification_metric.recall_score\n\n mlflow.log_metric(\"f1_score\",f1_score)\n mlflow.log_metric(\"precision\",precision_score)\n mlflow.log_metric(\"recall_score\",recall_score)\n mlflow.sklearn.log_model(best_model,\"model\", input_example=input_example)\n except Exception as e:\n raise MachinePredictiveMaintenanceException(e, sys)\n \n def train_model(self, X_train, y_train, X_test, y_test ):\n \"\"\"\n Train multiple models and select the best-performing one based on evaluation metrics.\n\n Args:\n X_train: Training features.\n y_train: Training labels.\n X_test: Testing features.\n y_test: Testing labels.\n\n Returns:\n ModelTrainerArtifact: An artifact containing details about the trained model and its metrics.\n \"\"\"\n models = {\n \"Random Forest\": RandomForestClassifier(verbose=1),\n \"Decision Tree\": DecisionTreeClassifier(),\n \"Gradient Boosting\": GradientBoostingClassifier(verbose=1),\n \"Logistic Regression\": LogisticRegression(verbose=1),\n \"AdaBoost\": AdaBoostClassifier(),\n }\n params={\n \"Decision Tree\": {\n 'criterion':['gini', 'entropy', 'log_loss'],\n # 'splitter':['best','random'],\n # 'max_features':['sqrt','log2'],\n },\n \"Random Forest\":{\n # 'criterion':['gini', 'entropy', 'log_loss'],\n \n # 'max_features':['sqrt','log2',None],\n 'n_estimators': [8,16,32,128,256]\n },\n \"Gradient Boosting\":{\n # 'loss':['log_loss', 'exponential'],\n 'learning_rate':[.1,.01,.05,.001],\n 'subsample':[0.6,0.7,0.75,0.85,0.9],\n # 'criterion':['squared_error', 'friedman_mse'],\n # 'max_features':['auto','sqrt','log2'],\n 'n_estimators': [8,16,32,64,128,256]\n },\n \"Logistic Regression\":{},\n \"AdaBoost\":{\n 'learning_rate':[.1,.01,.001],\n 'n_estimators': [8,16,32,64,128,256]\n }\n \n }\n model_report: dict = evaluate_models(X_train=X_train, y_train=y_train, X_test=X_test, y_test=y_test,\n models=models, param=params)\n best_model_score = max(sorted(model_report.values()))\n logging.info(f\"Best Model Score: {best_model_score}\")\n best_model_name = list(model_report.keys())[\n list(model_report.values()).index(best_model_score)\n ]\n logging.info(f\"Best Model Name: {best_model_name}\")\n best_model = models[best_model_name]\n\n y_train_pred = best_model.predict(X_train)\n classification_train_metric = get_classification_score(y_true=y_train, y_pred=y_train_pred)\n\n input_example = X_train[:1]\n print(input_example)\n # Track the experiments with mlflow\n self.track_mlflow(best_model, classification_train_metric, input_example)\n\n\n y_test_pred=best_model.predict(X_test)\n classification_test_metric = get_classification_score(y_true=y_test, y_pred=y_test_pred)\n\n # Track the experiments with mlflow\n self.track_mlflow(best_model, classification_test_metric, input_example)\n\n\n preprocessor = load_object(file_path=self.data_transformation_artifact.transformed_object_file_path)\n\n\n model_dir_path = os.path.dirname(self.model_trainer_config.trained_model_file_path)\n os.makedirs(model_dir_path,exist_ok=True)\n\n Machine_Predictive_Model = MachinePredictiveModel(model=best_model)\n\n save_object(self.model_trainer_config.trained_model_file_path,obj=MachinePredictiveModel)\n\n save_object(\"final_model\/model.pkl\",best_model)\n\n ## Model Trainer Artifact\n model_trainer_artifact=ModelTrainerArtifact(trained_model_file_path=self.model_trainer_config.trained_model_file_path,\n train_metric_artifact=classification_train_metric,\n test_metric_artifact=classification_test_metric\n )\n logging.info(f\"Model trainer artifact: {model_trainer_artifact}\")\n return model_trainer_artifact\n\n def initiate_model_trainer(self) -> ModelTrainerArtifact:\n\n try:\n train_file_path = self.data_transformation_artifact.transformed_train_file_path\n test_file_path = self.data_transformation_artifact.transformed_test_file_path\n\n #loading training array and testing array\n train_arr = load_numpy_array_data(train_file_path)\n test_arr = load_numpy_array_data(test_file_path)\n\n X_train, y_train, X_test, y_test = (\n train_arr[:, :-1],\n train_arr[:, -1],\n test_arr[:, :-1],\n test_arr[:, -1],\n )\n\n model_trainer_artifact=self.train_model(X_train,y_train,X_test,y_test)\n return model_trainer_artifact\n \n except Exception as e:\n raise MachinePredictiveMaintenanceException(e, sys)<\/code><\/pre>\ndef run_pipeline(self):\n\n \"\"\"\n Executes the entire training pipeline.\n\n Returns:\n ModelTrainerArtifact: Contains metadata about the trained model.\n \"\"\"\n try:\n data_ingestion_artifact= self.data_ingestion()\n data_validation_artifact= self.data_validation(data_ingestion_artifact=data_ingestion_artifact)\n data_transformation_artifact= self.data_transformation(data_validation_artifact=data_validation_artifact)\n model_trainer_artifact= self.model_trainer(data_transformation_artifact=data_transformation_artifact)\n self.sync_artifact_dir_to_s3()\n self.sync_saved_model_dir_to_s3()\n \n return model_trainer_artifact\n except Exception as e:\n raise MachinePredictiveMaintenanceException(e,sys)<\/code><\/pre>\n![\"training_pipeline\"](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2025\/02\/training_pipeline.webp\")
<\/figure>\npython training_pipeline.py<\/code><\/pre>\nmlflow ui # Launch the MLflow dashboard to monitor experiments.<\/code><\/pre>\n![\"ML](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2025\/02\/mlflow_pwrlSun.webp\")
<\/figure>\nAWS Integration<\/h2>\n
![\"AWS](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2025\/02\/AWS-Integration.webp\")
<\/figure>\nAWS S3<\/h2>\n
Step1: Download AWS CLI<\/h3>\n
\n
Step2:\u00a0Log in to AWS IAM User<\/h3>\n
\n
Step3: Navigate to S3 Service<\/h3>\n
\n
def sync_artifact_dir_to_s3(self):\n\n \"\"\"\n Syncs the artifact directory to S3.\n \"\"\"\n try:\n\n aws_bucket_url = f\"s3:\/\/{TRAINING_BUCKET_NAME}\/artifact\/{self.training_pipeline_config.timestamp}\"\n self.s3_sync.sync_folder_to_s3(folder = self.training_pipeline_config.artifact_dir,aws_bucket_url=aws_bucket_url)\n\n except Exception as e:\n raise MachinePredictiveMaintenanceException(e,sys)\n \n def sync_saved_model_dir_to_s3(self):\n\n \"\"\"\n Syncs the saved model directory to S3.\n \"\"\"\n\n try:\n aws_bucket_url = f\"s3:\/\/{TRAINING_BUCKET_NAME}\/final_model\/{self.training_pipeline_config.timestamp}\"\n self.s3_sync.sync_folder_to_s3(folder = self.training_pipeline_config.model_dir,aws_bucket_url=aws_bucket_url)\n except Exception as e:\n raise MachinePredictiveMaintenanceException(e,sys)<\/code><\/pre>\n![\"aws_s3_bucket\"](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2025\/02\/aws_s3_bucket.webp\")
<\/figure>\nAmazon Elastic Container Registry (ECR)<\/h2>\n
\n
![\"Amazon](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2025\/02\/Amazon-Elastic-Container-Registry-ECR.webp\")
<\/figure>\nDocker Integration for Deployment<\/h2>\n
FROM python:3.10-slim-buster\nWORKDIR \/app\nCOPY . \/app\n\nRUN apt update -y && apt install awscli -y\n\nRUN apt-get update && pip install -r requirements.txt\nCMD [\"python3\", \"app.py\"]<\/code><\/pre>\nSetup Action Secrets and Variables\u00a0<\/h2>\n
Step1: Open Your GitHub Repository<\/h4>\n
Step2:\u00a0Go to Settings<\/h4>\n
Step3: Access Secrets and Variables<\/h4>\n
Step4:\u00a0Create a New Secret<\/h4>\n
Step5:\u00a0Add AWS Credentials<\/h4>\n
\n
Step6:\u00a0Add AWS ECR Repository URI<\/h4>\n
\n
![\"\"](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2025\/02\/Screenshot_2025-02-09_194621.webp\")
<\/figure>\nAWS EC2\u00a0<\/h2>\n
\n
AWS EC2 CLI<\/h3>\n
sudo apt-get update -y\nsudo apt-get upgrade\ncurl -fsSL https:\/\/get.docker.com -o get-docker.sh\nsudo sh get-docker.sh\nsudo usermod -aG docker ubuntu\nnewgrp docker<\/code><\/pre>\n# Create a folder and navigate into it\nmkdir actions-runner && cd actions-runner\n\n# Download the latest runner package\ncurl -o actions-runner-linux-x64-2.322.0.tar.gz -L https:\/\/github.com\/actions\/runner\/releases\/download\/v2.322.0\/actions-runner-linux-x64-2.322.0.tar.gz\n\n# Optional: Validate the hash\necho \"b13b784808359f31bc79b08a191f5f83757852957dd8fe3dbfcc38202ccf5768 actions-runner-linux-x64-2.322.0.tar.gz\" | shasum -a 256 -c\n\n# Extract the installer\ntar xzf .\/actions-runner-linux-x64-2.322.0.tar.gz\n\n# Configure the runner\n.\/config.sh --url https:\/\/github.com\/karthikponna\/Predictive_Maintenance_MLOps --token \"Paste your token here\"\n\n# Run the runner\n.\/run.sh<\/code><\/pre>\nCI\/CD with GitHub Actions<\/h2>\n
\n
![\"GitHub](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2025\/02\/GitHub-Actions.webp\")
<\/figure>\n![\"aws_ec2_cli_1\"](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2025\/02\/aws_ec2_cli_1.webp\")
<\/figure>\nFastAPI<\/h2>\n
![\"FastAPI\"](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2025\/02\/fast_api_interface.webp\")
<\/figure>\nTrain route<\/h3>\n
![\"](\"https:\/\/av-eks-lekhak.s3.amazonaws.com\/media\/article_images\/train_route.png\")
<\/figure>\nPredict route<\/h4>\n
![\"](\"https:\/\/av-eks-lekhak.s3.amazonaws.com\/media\/article_images\/predict_route.png\")
Conclusion<\/h2>\n
![\"Karthik](\"https:\/\/av-eks-lekhak.s3.amazonaws.com\/media\/lekhak-profile-images\/converted_image_3tjdORM.webp\")
<\/p>\n