Machines don\u2019t break down out of nowhere\u2014there are at all times indicators. The issue? People aren\u2019t at all times nice at noticing them. That\u2019s the place Machine Predictive Upkeep is available in! This information will take you thru the thrilling world of Machine Predictive Upkeep, utilizing AWS and MLOps <\/a>to make sure your gear stays predictably dependable.<\/p>\n This text was printed as part of the\u00a0<\/strong><\/em>Information Science Blogathon.<\/strong><\/em><\/a><\/p>\n Sudden gear failures in industrial settings trigger downtime and monetary losses. In our mission, we’re utilizing MLOps greatest practices and machine studying to detect points early, enabling well timed repairs and decreasing disruptions.<\/p>\n Earlier than diving into implementation, let\u2019s take a better take a look at the mission structure.<\/p>\n Beneath we’ll first look within the conditions required:<\/p>\n Clone the repository<\/strong>:<\/p>\n Create and activate the digital setting<\/strong>:<\/p>\n Set up Required Dependencies<\/strong>:<\/p>\n Set Up Surroundings Variables:<\/strong><\/p>\n The Undertaking Construction outlines the important thing parts and group of the mission, guaranteeing readability and maintainability. It helps in understanding how completely different modules work together and the way the general system is designed. A well-defined construction simplifies growth, debugging, and scalability.<\/p>\n On this mission, we use a machine predictive upkeep CSV file, changing it into JSON information, and inserting it right into a MongoDB assortment.\u00a0<\/p>\n Dataset Hyperlink: https:\/\/www.kaggle.com\/datasets\/shivamb\/machine-predictive-maintenance-classification<\/a><\/p>\n Right here is the code snippet to transform CSV -> JSON information -> MongoDB<\/p>\n Right here is the code snippet to fetch information from MongoDB, cut up the info into practice and check CSV information, and retailer them as a Information Ingestion artifact.<\/p>\n On this step, we\u2019ll verify if the ingested information meets the anticipated format by guaranteeing all required columns are current utilizing a predefined schema and evaluating the coaching and testing information for any variations. Then we save the clear information and create a drift report, so solely high quality information is used for the subsequent step, which is remodeling the info.\u00a0<\/p>\n Information Validation code snippet:<\/p>\n Beneath is the drift report generated by the Information Validation.<\/p>\n Right here, we’ll clear and rework the validated information by changing temperature options from Kelvin to Celsius, dropping pointless columns, and making use of a change pipeline that makes use of ordinal encoding and Min-Max scaling, and dealing with information imbalance utilizing SMOTEENN.<\/p>\n The reworked coaching and check datasets are saved as .npy information<\/b> together with a serialized preprocessing object i.e. Minmax scaler(preprocessing.pkl<\/b>) all encapsulated as an artifact for additional mannequin coaching.<\/p>\n Now we’ll practice completely different classification fashions on the reworked coaching information like Resolution Tree,\u00a0Random Forest,\u00a0Gradient Boosting,\u00a0Logistic Regression, and AdaBoost fashions and consider their efficiency utilizing analysis metrics like f1-score, precision, and recall, after which log these particulars with MLflow<\/a>.<\/p>\n After selecting the right mannequin, it saves each the mannequin and transformation object domestically to artifact and analysis metrics to make use of it for deployment.<\/p>\n Now, let\u2019s create a training_pipeline.py<\/b> file the place we sequentially combine all of the steps of information ingestion, validation, transformation, and mannequin coaching into a whole pipeline.<\/p>\n You possibly can visually see how we have now constructed the training_pipeline<\/a>.<\/p>\n Now that we\u2019ve accomplished creating the pipeline, run the training_pipeline.py file to view the artifacts generated within the earlier steps.<\/p>\n Run the next command to view the MLflow dashboard.<\/p>\n As you may see, we have now efficiently logged mannequin metrics like recall, precision, and F1-score in MLflow.<\/p>\n After operating the training_pipeline.py file, we generated artifacts and saved them domestically. Now, we’ll retailer these artifacts in an AWS S3 bucket to allow cloud-based storage and accessibility.<\/p>\n Utilizing a Docker picture, we’ll push it to AWS ECR by way of GitHub Actions after which deploy it to manufacturing utilizing AWS EC2. We are going to focus on this course of in additional element within the upcoming sections.<\/p>\n Observe these steps to create an AWS S3 bucket:<\/p>\n Your IAM person has been efficiently linked to your mission.<\/p>\n Your bucket is now created, and you can begin importing artifacts. Now, add the next code to your training_pipeline.py file and run it once more to see the artifacts in your AWS S3 bucket.<\/p>\n Observe these steps to create an ECR repository.<\/p>\n Copy the URI out of your ECR repository, which ought to look one thing like \u201c788614365622.dkr.ecr.ap-southeast-2.amazonaws.com<\/b>\u201c, and reserve it someplace. We are going to want it later to stick into GitHub Secrets and techniques.<\/p>\n Dockerizing our mission ensures it runs easily in any setting with out dependency points. It\u2019s a must have instrument for packaging and sharing functions effortlessly.<\/p>\n Holding the Docker picture as small as attainable is essential for effectivity. We will reduce its dimension through the use of strategies like multi-staging and selecting a light-weight Python base picture.<\/p>\n To securely retailer delicate info like AWS credentials and repository URIs, we have to arrange GitHub Motion Secrets and techniques in our repository. Observe these steps:<\/p>\n Navigate to your repository on GitHub.<\/p>\n On the high of your repository web page, find and click on on the \u201cSettings\u201d tab.<\/p>\n Within the left sidebar, scroll all the way down to Secrets and techniques and Variables \u2192 Choose Actions.<\/p>\n Click on the New repository secret button.<\/p>\n Now let\u2019s perceive find out how to create an occasion with AWS EC2. Observe these steps.<\/p>\n After creating the occasion named machinepredictive, choose its Occasion ID and click on Join. Then, underneath EC2 Occasion Join, click on Join once more. This can open an AWS CLI interface the place you may run the required instructions.<\/p>\n Now, enter these instructions within the CLI one after the other to arrange Docker in your EC2 occasion:<\/p>\n Hold your CLI session open. Subsequent, navigate to your GitHub repository, click on on Settings, then go to the Actions part, and underneath Runners, click on New self-hosted runner. Select the Linux runner picture. Now, run the next instructions one after the other in your CLI to obtain and configure your self-hosted runner:<\/p>\n This can obtain, configure, and begin your self-hosted GitHub Actions runner in your EC2 occasion.<\/p>\n After establishing the self-hosted GitHub Actions runner, the CLI will immediate you to enter a reputation for the runner. Sort self-hosted and press Enter. <\/p>\n You possibly can try the .github\/workflows\/predominant.yml<\/b> code<\/a>\u00a0file.<\/p>\n Now, let\u2019s dive into what every part of this predominant.yml file does.<\/p>\n As soon as the Steady Deployment Job completes efficiently, you\u2019ll see an output like this within the AWS CLI.<\/p>\n Open your AWS EC2 occasion by clicking its Occasion ID. Confirm that the occasion state is Operating and find the Public IPv4 DNS. Click on on the \u201cOpen handle\u201d choice to robotically launch your FastAPI utility in your browser. Now, let\u2019s dive into FastAPI.<\/p>\n You possibly can try the app.py<\/b> file<\/a>.\u00a0<\/p>\n On this FastAPI utility, I\u2019ve created two predominant routes, one for coaching the mannequin and one other for producing predictions. Let\u2019s discover every route intimately.\u00a0\u00a0<\/p>\n The \/practice endpoint begins the mannequin coaching course of utilizing a predefined dataset, making it simple to replace and enhance the mannequin with new information.\u00a0It’s particularly helpful for retraining the mannequin to enhance accuracy or incorporate new information, guaranteeing the mannequin stays up-to-date and performs optimally.<\/p>\n The \/predict endpoint accepts a CSV file by way of POST. This endpoint handles incoming information, leverages the educated mannequin to generate predictions, after which returns the outcomes formatted as JSON.\u00a0This route is ideal for making use of the mannequin to new datasets, making it environment friendly for large-scale prediction duties.<\/p>\n Within the \/predict route, we\u2019ve included a pattern check.csv file; you may obtain it right here<\/a>.<\/p>\n Collectively, we constructed a full production-ready Predictive Upkeep MLOps mission\u2014from gathering and preprocessing information to coaching, evaluating, and deploying our mannequin utilizing Docker, AWS, and FastAPI. This mission reveals how MLOps can bridge the hole between growth and manufacturing, making it simpler to construct sturdy, scalable options.<\/p>\n Bear in mind, this information is all about studying and making use of these strategies to your information science tasks. Don\u2019t hesitate to experiment, innovate, and discover new enhancements as you progress. Thanks for sticking with me till the top\u2014continue to learn, hold doing, and continue to grow!<\/p>\nStudying Aims<\/h3>\n
\n
Downside: Unplanned Downtime & Upkeep Prices<\/h2>\n
![\"machine_Predictive_Maintenance\"](\"https:\/\/cdn.analyticsvidhya.com\/wp-content\/uploads\/2025\/02\/machine_Predictive_Maintenance.webp\")
<\/figure>\nEssential Conditions<\/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\nsupply venv\/bin\/activate\n\n# For Home windows:\npython -m venv venv\n.venvScriptsactivate<\/code><\/pre>\npip set up -r necessities.txt<\/code><\/pre>\n# Create a `.env` file and add your MongoDB connection string:\nMONGO_URI=your_mongodb_connection_string<\/code><\/pre>\nUndertaking Construction<\/h2>\n
project_root\/\n\u2502\n\u251c\u2500\u2500 .github\/\n\u2502 \u2514\u2500\u2500 workflows\/\n\u2502 \u2514\u2500\u2500 predominant.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 mannequin.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 parts\/\n\u2502 \u251c\u2500\u2500 fixed\/\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 desk.html\n\u2502\n\u251c\u2500\u2500 valid_data\/\n| \u2514\u2500\u2500check.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 predominant.py\n\u251c\u2500\u2500 push_data.py\n\u251c\u2500\u2500 README.md\n\u251c\u2500\u2500 necessities.txt\n\u251c\u2500\u2500 setup.py\n\u251c\u2500\u2500 test_mongodb.py<\/code><\/pre>\nInformation Ingestion<\/h2>\n
class PredictiveDataExtract():\n\n def __init__(self):\n \"\"\"\n Initializes the PredictiveDataExtract class.\n \"\"\"\n attempt:\n cross\n besides Exception as e:\n increase MachinePredictiveMaintenanceException(e, sys)\n \n def csv_to_json_convertor(self, file_path):\n attempt:\n information = pd.read_csv(file_path)\n information.reset_index(drop=True, inplace=True)\n information = listing(json.hundreds(information.T.to_json()).values())\n return information\n besides Exception as e:\n increase MachinePredictiveMaintenanceException(e, sys)\n \n def insert_data_mongodb(self, information, database, assortment):\n attempt:\n self.information = information\n self.database = database\n self.assortment = assortment\n\n self.mongo_client = pymongo.MongoClient(MONGO_DB_URL)\n self.database = self.mongo_client[self.database]\n \n self.assortment = self.database[self.collection]\n self.assortment.insert_many(self.information)\n return(len(self.information))\n besides Exception as e:\n increase MachinePredictiveMaintenanceException(e, sys)\n \nif __name__==\"__main__\":\n FILE_PATH=\"Machine_Predictive_Data\/predictive_maintenance.csv\"\n DATABASE=\"Predictive_Maintenance_MLOps\"\n assortment = \"Machine_Predictive_Data\"\n\n predictive_data_obj=PredictiveDataExtract()\n information = predictive_data_obj.csv_to_json_convertor(FILE_PATH)\n no_of_records = predictive_data_obj.insert_data_mongodb(information, DATABASE, assortment)\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 supplied configuration.\n\n Parameters:\n data_ingestion_config: DataIngestionConfig\n Configuration object containing particulars for information ingestion.\n \"\"\"\n attempt:\n self.data_ingestion_config=data_ingestion_config\n besides Exception as e:\n increase MachinePredictiveMaintenanceException(e, sys)\n \n def export_collection_as_dataframe(self):\n attempt:\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 assortment = self.mongo_client[database_name][collection_name]\n df = pd.DataFrame(listing(assortment.discover()))\n if \"_id\" in df.columns.to_list():\n df = df.drop(columns=[\"_id\"], axis=1)\n df.substitute({\"na\":np.nan},inplace=True)\n return df\n besides Exception as e:\n increase MachinePredictiveMaintenanceException(e, sys)\n\n def export_data_into_feature_store(self,dataframe: pd.DataFrame):\n attempt:\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 besides Exception as e:\n increase MachinePredictiveMaintenanceException(e,sys)\n\n def split_data_as_train_test(self, dataframe:pd.DataFrame): \n attempt:\n train_set, test_set = train_test_split(\n dataframe, test_size=self.data_ingestion_config.train_test_split_ratio\n )\n logging.data(\"Carried out practice check cut up on the dataframe\")\n logging.data(\n \"Exited split_data_as_train_test methodology 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.data(f\"Exporting practice and check 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.data(f\"Exported practice and check file path.\" )\n besides Exception as e:\n increase MachinePredictiveMaintenanceException(e, sys)\n\n def initiate_data_ingestion(self):\n attempt:\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 besides Exception as e:\n increase MachinePredictiveMaintenanceException(e,sys)<\/code><\/pre>\nInformation Validation<\/h2>\n
class DataValidation:\n\n def __init__(self, data_ingestion_artifact:DataIngestionArtifact,\n data_validation_config: DataValidationConfig):\n attempt:\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 besides Exception as e:\n increase MachinePredictiveMaintenanceException(e, sys)\n \n @staticmethod\n def read_data(file_path) -> pd.DataFrame:\n attempt:\n return pd.read_csv(file_path)\n besides Exception as e:\n increase MachinePredictiveMaintenanceException(e, sys) \n\n \n def validate_number_of_columns(self, dataframe:pd.DataFrame)-> bool:\n attempt:\n number_of_columns = len(self._schema_config)\n logging.data(f\"Required variety of columns:{number_of_columns}\")\n logging.data(f\"Information body has columns:{len(dataframe.columns)}\")\n\n if len(dataframe.columns) == number_of_columns: \n return True\n return False\n besides Exception as e:\n increase MachinePredictiveMaintenanceException (e, sys)\n \n\n def is_columns_exist(self, df:pd.DataFrame) -> bool:\n attempt:\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.data(f\"Lacking 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.data(f\"Lacking categorical column: {missing_categorical_columns}\")\n return False if len(missing_categorical_columns)>0 or len(missing_numerical_columns)>0 else True\n \n besides Exception as e:\n increase MachinePredictiveMaintenanceException(e, sys)\n\n \n def detect_dataset_drift(self, base_df, current_df, threshold = 0.05) -> bool :\n attempt:\n standing = 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 <= is_same_dist.pvalue:\n is_found = False\n\n else: \n is_found = True\n standing = False\n\n report.replace({column:{\n \"p_value\": float(is_same_dist.pvalue),\n \"drift_status\": is_found\n }})\n drift_report_file_path = self.data_validation_config.drift_report_file_path\n dir_path = os.path.dirname(drift_report_file_path)# create listing\n os.makedirs(dir_path, exist_ok=True)\n\n write_yaml_file(file_path=drift_report_file_path, content material=report)\n besides Exception as e:\n increase MachinePredictiveMaintenanceException(e, sys)\n \n def initiate_data_validation(self)-> DataValidationArtifact:\n attempt:\n validation_error_msg = \"\"\n logging.data(\"Beginning information 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 # learn the info from the practice and check \n train_dataframe = DataValidation.read_data(train_file_path)\n test_dataframe = DataValidation.read_data(test_file_path)\n\n # validate variety of columns\n standing = self.validate_number_of_columns(dataframe=train_dataframe)\n logging.data(f\"All required columns current in coaching dataframe: {standing}\")\n\n if not standing:\n validation_error_msg += f\"Prepare dataframe doesn't include all columns.n\"\n\n standing = self.validate_number_of_columns(dataframe=test_dataframe)\n\n if not standing:\n validation_error_msg += f\"Take a look at dataframe doesn't include all columns.n\"\n\n standing = self.is_columns_exist(df=train_dataframe)\n\n if not standing:\n validation_error_msg += f\"Columns are lacking in coaching dataframe.\"\n\n standing = self.is_columns_exist(df=test_dataframe)\n\n if not standing:\n validation_error_msg += f\"columns are lacking in check dataframe.\"\n ## lets verify datadrift\n standing=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=standing,\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 besides Exception as e:\n increase MachinePredictiveMaintenanceException(e,sys)<\/code><\/pre>\nAir temperature [K]:\n drift_status: true\n p_value: 0.016622943467175914\nFailure Sort:\n drift_status: false\n p_value: 1.0\nCourse of temperature [K]:\n drift_status: false\n p_value: 0.052940072765804994\nProduct ID:\n drift_status: false\n p_value: 0.09120557172716418\nRotational pace [rpm]:\n drift_status: false\n p_value: 0.2673520066245566\nGoal:\n drift_status: false\n p_value: 0.999999998717466\nInstrument put on [min]:\n drift_status: false\n p_value: 0.13090856779628832\nTorque [Nm]:\n drift_status: false\n p_value: 0.5001773464540389\nSort:\n drift_status: false\n p_value: 1.0\nUDI:\n drift_status: true\n p_value: 0.022542489133976953<\/code><\/pre>\nInformation Transformation<\/h2>\n
class DataTransformation:\n\n def __init__(self,data_validation_artifact: DataValidationArtifact,\n data_transformation_config: DataTransformationConfig):\n attempt:\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 besides Exception as e:\n increase MachinePredictiveMaintenanceException(e,sys)\n \n @staticmethod\n def read_data(file_path) -> pd.DataFrame:\n attempt:\n return pd.read_csv(file_path)\n besides Exception as e:\n increase MachinePredictiveMaintenanceException(e, sys)\n \n def get_data_transformer_object(self):\n attempt:\n logging.data(\"Bought numerical cols from schema config\")\n \n scaler = MinMaxScaler()\n # Fetching classes for OrdinalEncoder from schema config\n ordinal_categories = self._schema_config.get('ordinal_categories', [])\n ordinal_encoder = OrdinalEncoder(classes=ordinal_categories)\n\n logging.data(\"Initialized MinMaxScaler, OrdinalEncoder with classes\")\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.data(\"Created preprocessor object from ColumnTransformer\")\n return preprocessor\n besides Exception as e:\n increase MachinePredictiveMaintenanceException(e, sys)\n \n def initiate_data_transformation(self) -> DataTransformationArtifact:\n attempt:\n \n logging.data(\"Beginning information 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.data(\"Accomplished dropping the columns for Coaching 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.data(\"Accomplished dropping the columns for Testing dataset\")\n input_feature_train_arr = preprocessor.fit_transform(input_feature_train_df)\n input_feature_test_arr = preprocessor.rework(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.data(\"Utilized SMOTEENN on coaching 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 besides Exception as e:\n increase MachinePredictiveMaintenanceException(e, sys)<\/code><\/pre>\nCoaching and Analysis<\/h2>\n
class ModelTrainer:\n def __init__(self, data_transformation_artifact: DataTransformationArtifact, model_trainer_config: ModelTrainerConfig):\n attempt:\n self.data_transformation_artifact = data_transformation_artifact\n self.model_trainer_config = model_trainer_config\n\n besides Exception as e:\n increase MachinePredictiveMaintenanceException(e, sys)\n \n def track_mlflow(self, best_model, classification_metric, input_example):\n \"\"\"\n Log mannequin and metrics to MLflow.\n\n Args:\n best_model: The educated mannequin object.\n classification_metric: The classification metrics (f1, precision, recall).\n input_example: An instance enter information pattern for the mannequin.\n \"\"\"\n attempt:\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,\"mannequin\", input_example=input_example)\n besides Exception as e:\n increase MachinePredictiveMaintenanceException(e, sys)\n \n def train_model(self, X_train, y_train, X_test, y_test ):\n \"\"\"\n Prepare a number of fashions and choose the best-performing one primarily based on analysis metrics.\n\n Args:\n X_train: Coaching options.\n y_train: Coaching labels.\n X_test: Testing options.\n y_test: Testing labels.\n\n Returns:\n ModelTrainerArtifact: An artifact containing particulars in regards to the educated mannequin and its metrics.\n \"\"\"\n fashions = {\n \"Random Forest\": RandomForestClassifier(verbose=1),\n \"Resolution Tree\": DecisionTreeClassifier(),\n \"Gradient Boosting\": GradientBoostingClassifier(verbose=1),\n \"Logistic Regression\": LogisticRegression(verbose=1),\n \"AdaBoost\": AdaBoostClassifier(),\n }\n params={\n \"Resolution 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 fashions=fashions, param=params)\n best_model_score = max(sorted(model_report.values()))\n logging.data(f\"Greatest Mannequin Rating: {best_model_score}\")\n best_model_name = listing(model_report.keys())[\n list(model_report.values()).index(best_model_score)\n ]\n logging.data(f\"Greatest Mannequin Identify: {best_model_name}\")\n best_model = fashions[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 # Observe 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 # Observe 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(mannequin=best_model)\n\n save_object(self.model_trainer_config.trained_model_file_path,obj=MachinePredictiveModel)\n\n save_object(\"final_model\/mannequin.pkl\",best_model)\n\n ## Mannequin Coach 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.data(f\"Mannequin coach artifact: {model_trainer_artifact}\")\n return model_trainer_artifact\n\n def initiate_model_trainer(self) -> ModelTrainerArtifact:\n\n attempt:\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 coaching 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 besides Exception as e:\n increase MachinePredictiveMaintenanceException(e, sys)<\/code><\/pre>\ndef run_pipeline(self):\n\n \"\"\"\n Executes your complete coaching pipeline.\n\n Returns:\n ModelTrainerArtifact: Incorporates metadata in regards to the educated mannequin.\n \"\"\"\n attempt:\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 besides Exception as e:\n increase 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 watch 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: Obtain AWS CLI<\/h3>\n
\n
Step2:\u00a0Log in to AWS IAM Person<\/h3>\n
\n
Step3: Navigate to S3 Service<\/h3>\n
\n
def sync_artifact_dir_to_s3(self):\n\n \"\"\"\n Syncs the artifact listing to S3.\n \"\"\"\n attempt:\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 besides Exception as e:\n increase MachinePredictiveMaintenanceException(e,sys)\n \n def sync_saved_model_dir_to_s3(self):\n\n \"\"\"\n Syncs the saved mannequin listing to S3.\n \"\"\"\n\n attempt:\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 besides Exception as e:\n increase 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 replace -y && apt set up awscli -y\n\nRUN apt-get replace && pip set up -r necessities.txt\nCMD [\"python3\", \"app.py\"]<\/code><\/pre>\nSetup Motion Secrets and techniques and Variables\u00a0<\/h2>\n
Step1: Open Your GitHub Repository<\/h4>\n
Step2:\u00a0Go to Settings<\/h4>\n
Step3: Entry Secrets and techniques 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 replace -y\nsudo apt-get improve\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# Obtain the most recent runner package deal\ncurl -o actions-runner-linux-x64-2.322.0.tar.gz -L https:\/\/github.com\/actions\/runner\/releases\/obtain\/v2.322.0\/actions-runner-linux-x64-2.322.0.tar.gz\n\n# Optionally available: 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 right 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>\nPrepare 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