Worker Attrition Charge Prediction Utilizing ZenML and Streamlit

Introduction

Are u working as an HR ? struggling to foretell whether or not the workers in your group will proceed working or they’re think about leaving the organisation, No worries ! you don’t wanna be a astrologer to foretell this, by utilizing the facility of Knowledge Science, we will predict it precisely. Allow us to start our great journey of worker Attrition charge with a easy, but highly effective MLOps device, known as ZenML and streamlit. Let’s begin our journey.

Studying Targets

On this article, we’ll be taught,

• What’s ZenML? Why and The best way to use it?
• Why to make use of MLflow and combine with ZenML?
• The necessity of utilizing deployment pipeline
• Implementation of worker attrition charge undertaking and make predictions

This text was printed as part of the Data Science Blogathon.

Mission Implementation

Downside Assertion: Predict whether or not an worker will go away an organisation or not based mostly on a number of components like Age, revenue, Efficiency and so forth.,

Resolution: Construct a Logistic Regression mannequin to foretell the attrition charge of an Worker

Dataset: IBM HR Analytics Worker Attrition & Efficiency

[Source]: https://www.kaggle.com/datasets/pavansubhasht/ibm-hr-analytics-attrition-dataset

Earlier than seeing the implementation of our undertaking, allow us to see why we’re utilizing ZenML right here first.

Why ZenML?

ZenML is an easy and highly effective MLOps orchestration device used to create ML pipelines, cache  pipeline steps and save computational assets. ZenML additionally affords integration with a number of ML instruments, making it top-of-the-line device to create ML pipelines.We are able to monitor our mannequin steps,analysis metrics, we will see our pipelines visually in Dashboards and plenty of extra.

On this undertaking, we’ll implement a conventional pipeline, which makes use of ZenML, and we shall be integrating mlflow with zenml, for experiment monitoring.We can even implement a steady deployment pipeline utilizing MLflow integration with ZenML, which is able to ingest and clear the information, practice the mannequin and redeploys the mannequin, when the prediction meets some minimal analysis standards.With this pipeline, we will make certain, if any new mannequin performs higher than the earlier mannequin’s threshold prediction worth,then the MLFlow deployment server shall be up to date with the brand new mannequin as an alternative of the previous mannequin.

Widespread ZenML Phrases

• Pipelines: Sequence of steps in our Mission.
• Parts: Constructing blocks or a specific perform in our MLOps pipeline.
• Stacks: Assortment of elements in native/cloud.
• Artifacts: Enter and output information of a step,in our undertaking, which is saved in Artifact retailer.
• Artifact Retailer: Cupboard space for storing and model monitoring of our artifact.
• Materializers: Parts which defines how artifacts are saved and retrieved from the artifact retailer.
• Flavors: Options for particular instruments and use circumstances.
• ZenML Server: Deployment for working stack elements remotely.

Pre-requisites and Fundamental ZenML Instructions

• Activate your Digital Surroundings:

#create a digital setting
python3 -m venv venv
#Activate your digital environmnent in your undertaking folder
supply venv/bin/activate

All the essential ZenML Instructions with its functionalities are given under:

#Set up zenml
pip set up zenml

#to Launch zenml server and dashboard domestically
pip set up "zenml[server]"


#to see the zenml Model:
zenml model

#To provoke a brand new repository
zenml init

#to run the dashboard domestically:
zenml up


#to know the standing of our zenml Pipelines
zenml present

These instructions are essential to know to work with ZenML.

Integration of MLflow with ZenML

We’re utilizing mlflow because the experiment tracker, to trace our mannequin,artifacts, hyperparameter values. We’re registering the stack part, experiment tracker, model-deployer right here:

#Integrating mlflow with ZenML
zenml integration set up mlflow -y

#Register the experiment tracker
zenml experiment-tracker register mlflow_tracker_employee --flavor=mlflow

#Registering the mannequin deployer
zenml model-deployer register mlflow_employee --flavor=mlflow

#Registering the stack
zenml stack register mlflow_stack_employee -a default -o default -d mlflow_employee -e mlflow_tracker_employee --set

Zenml Stack Listing

Mission Construction

employee-attrition-prediction/          # Mission listing
├── information/                               
│   └── HR-Worker-Attrition.csv       # Dataset file
│
├── pipelines/                          
│   ├── deployment_pipeline.py          # Deployment pipeline
│   ├── training_pipeline.py            # Coaching pipeline
│   └── utils.py                        
│
├── src/                                # Supply code 
│   ├── data_cleaning.py                # Knowledge cleansing and preprocessing
│   ├── analysis.py                   # Mannequin analysis
│   └── model_dev.py                    # Mannequin improvement
│                 
├── steps/                              # code information for ZenML steps
│   ├── ingest_data.py                  # Ingestion of information
│   ├── clean_data.py                   # Knowledge cleansing and preprocessing
│   ├── model_train.py                  # Prepare the mannequin    
│   ├── analysis.py                   # Mannequin analysis
│   └── config.py                       
│
├── streamlit_app.py                    # Streamlit internet software
│
├── run_deployment.py                   # Code for working deployment and prediction pipeline
├── run_pipeline.py                     # Code for working coaching pipeline
│
├── necessities.txt                    # Listing of undertaking required packages
├── README.md                           # Mission documentation
└── .zen/                               # ZenML listing (created robotically after ZenML initialization)

Knowledge Ingestion

We first ingest the information from the HR-Worker-Attrition-Charge dataset from the information folder.

import pandas as pd
from zenml import step

class IngestData:
    def get_data(self) -> pd.DataFrame:
        df = pd.read_csv("./information/HR-Worker-Attrition.csv")
        return df

@step
def ingest_data() -> pd.DataFrame:
    ingest_data = IngestData()
    df = ingest_data.get_data()
    return df
    

@step is a decorator, used to make the perform ingest_data() as a step of the pipeline.

Exploratory Knowledge Evaluation

#Perceive the information
df.data()
# See how the information seems
df.describe()
# Test the pattern information
df.head()
#Test the null values
df.isnull.sum()

#Test the proportion of people that stayed and left the corporate:
df['Attrition'].value_counts()
df_left = df[df['Attrition'] == "Sure"]
df_stayed = df[df['Attrition'] == "No"]
left_percentage=df_left.form[0]*100/df.form[0]
stayed_percentage=df_stayed.form[0]*100/df.form[0]
print(f"The proportion of people that left the corporate are:{left_percentage}")
print(f"The proportion of people that stayed the corporate are:{stayed_percentage}")

#Analyse the variations in options between individuals who stayed and individuals who left the corporate
df_left.describe()
df_stayed.describe()

Output

Observations

• The workers who left the job have been labored much less years within the firm.
• The employess who left the corporate have been youthful, than the workers who stayed.
• The workers who left are having the workplace far distance from residence than stayed.

Knowledge Cleansing and Processing

• Knowledge Cleansing: Now we have eliminated the undesirable columns within the dataset resembling :”EmployeeCount”, “EmployeeNumber”, “StandardHours”, then we’ve got modified the options which have solely information values between Sure(or)No to binary 1(or)0.
• One sizzling Encoding: Then, we did one-hot encoding to the explicit columns resembling ‘BusinessTravel’, ‘Division’, ‘EducationField’, ‘Gender’, ‘JobRole’, ‘MaritalStatus’.

import pandas as pd
class DataPreProcessStrategy(DataStrategy):
    def __init__(self, encoder=None):
        self.encoder = encoder
    """This class is used to preprocess the given dataset"""
    def handle_data(self, information: pd.DataFrame) -> pd.DataFrame:
        strive:
            print("Column Names Earlier than Preprocessing:", information.columns)  # Add this line
            information = information.drop(["EmployeeCount", "EmployeeNumber", "StandardHours"], axis=1)
            if 'Attrition' in information.columns:
                print("Attrition column present in information.")
            else:
                print("Attrition column not present in information.")
            information["Attrition"] = information["Attrition"].apply(lambda x: 1 if x == "Sure" else 0)
            information["Over18"] = information["Over18"].apply(lambda x: 1 if x == "Sure" else 0)
            information["OverTime"] = information["OverTime"].apply(lambda x: 1 if x == "Sure" else 0)

            # Extract categorical variables
            cat = information[['BusinessTravel', 'Department', 'EducationField', 'Gender', 'JobRole', 'MaritalStatus']]

            # Carry out one-hot encoding on categorical variables
            onehot = OneHotEncoder()
            cat_encoded = onehot.fit_transform(cat).toarray()
            
            
            # Convert cat_encoded to DataFrame
            cat_df = pd.DataFrame(cat_encoded)

            # Extract numerical variables
            numerical = information[['Age', 'Attrition', 'DailyRate', 'DistanceFromHome', 'Education', 'EnvironmentSatisfaction', 'HourlyRate', 'JobInvolvement', 'JobLevel', 'JobSatisfaction', 'MonthlyIncome', 'MonthlyRate', 'NumCompaniesWorked', 'Over18', 'OverTime', 'PercentSalaryHike', 'PerformanceRating', 'RelationshipSatisfaction', 'StockOptionLevel', 'TotalWorkingYears', 'TrainingTimesLastYear', 'WorkLifeBalance', 'YearsAtCompany', 'YearsInCurrentRole', 'YearsSinceLastPromotion', 'YearsWithCurrManager']]

            # Concatenate X_cat_df and X_numerical
            information = pd.concat([cat_df, numerical], axis=1)

            print("Column Names After Preprocessing:", information.columns)  # Add this line
            print("Preprocessed Knowledge:")
            print(information.head())
            return information
        besides Exception as e:
            logging.error(f"Error in preprocessing the information: {e}")
            increase e
            

Output

The info seems like this, in spite of everything information cleansing and processing accomplished: You’ll be able to see within the picture lastly, the information consists of solely numerical information after encoding accomplished.

Splitting the Knowledge

We’ll then cut up the coaching and testing datasets within the ratio of 80:20.

from sklearn.model_selection import train_test_split
class DataDivideStrategy(DataStrategy):
    def handle_data(self, information: pd.DataFrame) -> Union[pd.DataFrame, pd.Series]:
        strive:
            # Test if 'Attrition' is current within the information
            if 'Attrition' in information.columns:
                X = information.drop(['Attrition'], axis=1)
                Y = information['Attrition']
                X_train, X_test, Y_train, Y_test = train_test_split(X, Y, test_size=0.2, random_state=42)
                return X_train, X_test, Y_train, Y_test
            else:
                increase ValueError("'Attrition' column not present in information.")
        besides Exception as e:
            logging.error(f"Error in information dealing with: {str(e)}")
            increase e
            

Mannequin Coaching

Since, its’s a Classification drawback, we’re utilizing Logistic Regression right here, we will additionally use Random forest Classifer, Gradient boosting and so forth., classification algorithms.

from zenml import pipeline
@training_pipeline
def training_pipeline(data_path: str):
    df = ingest_data(data_path)
    X_train, X_test, y_train, y_test = clean_and_split_data(df)
    mannequin = define_model()  # Outline your machine studying mannequin
    trained_model = train_model(mannequin, X_train, y_train)
    evaluation_metrics = evaluate_model(trained_model, X_test, y_test)

Right here, @training_pipeline decorator is used to defind the perform training_pipeline() as a pipeline in ZenML.

Analysis

For binary classification issues, we use analysis metrics resembling: accuracy, precision, F1 rating, ROC-AUC curve and so forth., We import classification_report from scikit-learn library, to calculate the analysis metrics and to provide us the classification report.
Code:

import logging
import numpy as np
from sklearn.metrics import classification_report

class ClassificationReport:
    @staticmethod
    def calculate_scores(y_true: np.ndarray, y_pred: np.ndarray):
        strive:
            logging.data("Calculate Classification Report")
            report = classification_report(y_true, y_pred, output_dict=True)
            logging.data(f"Classification Report:n{report}")
            return report
        besides Exception as e:
            logging.error(f"Error in calculating Classification Report: {e}")
            increase e

Classification Report:

To see the dashboard of the training_pipeline, we have to run the run_pipelilne.py,

run_pipelilne.py,

from zenml import pipeline
from pipelines.training_pipeline import train_pipeline
from zenml.shopper import Shopper
import pandas as pd

if __name__ == "__main__":
    uri = Shopper().active_stack.experiment_tracker.get_tracking_uri()
    print(uri)
    train_pipeline(data_path="./information/HR-Worker-Attrition.csv")

which is able to return the monitoring dashboard URL, seems like this,
“Dashboard URL: http://127.0.0.1:8237/workspaces/default/pipelines/6e7941f4-cf74-4e30-b3e3-ff4428b823d2/runs/2274fc18-aba1-4536-aaee-9d2ed7d26323/dag“
You’ll be able to click on the URL and consider your superb coaching pipeline in zenml dashboard. Right here, the entire pipeline picture is cut up into completely different picture components to see,it extra clearly intimately.

General the training_pipeline seems like this within the dashboard, given under:

Mannequin Deployment

Deployment Set off

class DeploymentTriggerConfig(BaseParameters):
    min_accuracy: float = 0.5

On this class DeploymentTriggerConfig, we set a minimal accuracy parameter, which specifies what our minimal mannequin accuracy needs to be.

Establishing Deployment Set off

@step(enable_cache=False)
def deployment_trigger(
    accuracy: float,
    config: DeploymentTriggerConfig,
):
    return accuracy > config.min_accuracy

Right here, this deployment_trigger() perform is used to deploy  the mannequin, solely when it exceeds the minimal accuracy. We’ll cowl about why we’ve got used caching right here within the subsequent part.

Steady Deployment Pipeline

@pipeline(enable_cache=False, settings={"docker":docker_settings})
def continuous_deployment_pipeline(
   data_path: str,
   #data_path="C:/Customers/person/Desktop/machine studying/Mission/zenml Pipeline/Customer_Satisfaction_project/information/olist_customers_dataset.csv",
   min_accuracy:float=0.0,
   employees: int=1,
   timeout: int=DEFAULT_SERVICE_START_STOP_TIMEOUT,
):
    
   df=ingest_data()
   # Clear the information and cut up into coaching/take a look at units
   X_train,X_test,Y_train,Y_test=clean_df(df)
   mannequin=train_model(X_train,X_test,Y_train,Y_test)
   evaluation_metrics=evaluate_model(mannequin,X_test,Y_test)
   deployment_decision=deployment_trigger(evaluation_metrics)    
   mlflow_model_deployer_step(
      mannequin=mannequin,
      deploy_decision=deployment_decision,
      employees=employees,
      timeout=timeout,
    )

Right here, on this continuous_deployment_pipeline(), we’ll ingest the information, clear the information, practice our mannequin, consider it, and deploy our mannequin provided that it passes the deployment_trigger() situation, in order that we will make certain the brand new mannequin we’re going to deploy, will execute provided that it’s prediction accuracy exceeds the earlier mannequin’s prediction accuracy,which is the brink worth. That is how the continous_deployment_pipeline() works.

 Caching refers to storing the output of the earlier executed steps within the pipeline. The outputs are saved within the Artifact retailer. We use  caching within the pipeline parameter, to say that, there isn’t a change within the outputs within the earlier runs and present working step, so zenML will reuse the earlier run output itself. Enabling caching will pace up the pipeline working course of and saves our computational assets. However typically, in conditions  the place, we have to run pipelines, the place there shall be dynamic change within the enter, parameters, output like our continuous_deployment_pipeline(), then turning off the caching is effectively and good. So, we’ve got written  enable_cache=False right here.

Inference Pipeline

We use inference pipeline to make predictions on the brand new information, based mostly on the deployed mannequin. Let’s see how we used this pipeline in our undertaking.

inference_pipeline()

@pipeline(enable_cache=False,settings={"docker":docker_settings})
def inference_pipeline(pipeline_name: str, pipeline_step_name:str):
   information=dynamic_importer()
   #print("Knowledge Form for Inference:", information.form)  # Print the form of information for inference
   service=prediction_service_loader(
      pipeline_name=pipeline_name,
      pipeline_step_name=pipeline_step_name,
      working=False,
      )
   prediction=predictor(service=service,information=information)
   return prediction
  

Right here, the inference_pipeline(), works within the following order:

• dynamic_importer()– First, the dynamic_importer() masses the brand new information and prepares it.
• prediction_service_loader()– The prediction_service_loader() masses the deployed mannequin, based mostly on the pipeline identify and step identify parameters.
• predictor()-Then, predictor() is used to foretell the brand new information based mostly on the deployed mannequin.

Allow us to see about every of those capabilities  under:

dynamic importer()

@step(enable_cache=False)
def dynamic_importer()->str:
   information=get_data_for_test()
   return information  

Right here, it calls the get_data_for_test() within the utils.py, which is able to masses the brand new information, do information processing and returns the information.

prediction_service_loader()

@step(enable_cache=False)
def prediction_service_loader(
   pipeline_name: str,
   pipeline_step_name: str,
   working:bool=True,
   model_name: str="mannequin", 
)->MLFlowDeploymentService:
   mlflow_model_deployer_component=MLFlowModelDeployer.get_active_model_deployer()
   existing_services=mlflow_model_deployer_component.find_model_server(
   pipeline_name=pipeline_name,
   pipeline_step_name=pipeline_step_name,
   model_name=model_name,
   working=working,
)
   if not existing_services:
      increase RuntimeError(
         f"No MLFlow deployment service discovered for pipeline {pipeline_name},step {pipeline_step_name} and mannequin{model_name} and pipeline for the mannequin {model_name} is at present working"
      )

Right here, on this prediction_service_loader (), we  load the deployment service with respect to the deployed mannequin based mostly on the parameters. A deployment service is a runtime setting, the place our deployed mannequin, is able to settle for inference requests to make predictions on the brand new information. The road existing_services=mlflow_model_deployer_component.find_model_server(), searches for any present deployment service obtainable based mostly on the given parameters like pipeline identify and pipeline step identify, if there isn’t a present companies obtainable, then it means the deployment pipeline will not be executed but, or there is a matter with the deployment pipeline, so it thows an Runtime Error.

predictor()

@step
def predictor(
    service: MLFlowDeploymentService,
    information: str,
) -> np.ndarray:
    
    """Run an inference request towards a prediction service"""

    service.begin(timeout=21)  # needs to be a NOP if already began
    information = json.masses(information)
    information.pop("columns")
    information.pop("index")
    columns_for_df = [
        0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,"Age","DailyRate","DistanceFromHome","Education","EnvironmentSatisfaction","HourlyRate","JobInvolvement","JobLevel","JobSatisfaction","MonthlyIncome","MonthlyRate","NumCompaniesWorked","Over18","OverTime","PercentSalaryHike","PerformanceRating","RelationshipSatisfaction","StockOptionLevel","TotalWorkingYears","TrainingTimesLastYear","WorkLifeBalance","YearsAtCompany","YearsInCurrentRole","YearsSinceLastPromotion","YearsWithCurrManager",
    ]
    df = pd.DataFrame(information["data"], columns=columns_for_df)
    json_list = json.masses(json.dumps(record(df.T.to_dict().values())))
    information = np.array(json_list)
    prediction = service.predict(information)
    return prediction

After, having the deployed mannequin and the brand new information, we will use the predictor(), to make the predictions.

To visually, see the continual deployment and inference pipeline, we have to run the run_deployment.py, the place the configurations, to deploy and predict shall be outlined.

@click on.possibility(
    "--config",
    sort=click on.Selection([DEPLOY, PREDICT, DEPLOY_AND_PREDICT]),
    default=DEPLOY_AND_PREDICT,
    assist="Optionally you'll be able to select to solely run the deployment "
    "pipeline to coach and deploy a mannequin (`deploy`), or to "
    "solely run a prediction towards the deployed mannequin "
    "(`predict`). By default each shall be run "
    "(`deploy_and_predict`).",
)

Right here, we will both run the continual deployment pipeline or the inference pipeline, by following these instructions,

#The continual deployment pipeline
python run_deployment.py

#To see the inference Pipeline(that's to deploy and predict)
python run_deployment.py --config predict

After executing, the instructions, you’ll be able to see the zenML dashboard URL,like this

Dashboard URL: http://127.0.0.1:8237/workspaces/default/pipelines/b437cf1a-971c-4a23-a3b6-c296c1cdf8ca/runs/58826e07-6139-453d-88f9-b3c771bb6695/dag

Take pleasure in your pipeline visualisations within the dashboard:

Steady deployment Pipeline

The continual deployment pipeline,(from ingestion of information to mlflow_model_deployer_step seems like),

Inference Pipeline

Constructing A Streamlit Software

Streamlit is a tremendous open-source, python based mostly framework, used to create  UI’s, we will use streamlit to construct internet apps shortly, with out realizing  backend or frontend improvement. First, we have to set up streamlit in our PC.
The instructions to put in and run streamlit server in our native system are,

#set up streamlit in our native PC
pip set up streamlit

#to run the streamlit native internet server
streamlit run streamlit_app.py

Code:

import json
import numpy as np
import pandas as pd
import streamlit as st
from PIL import Picture
from pipelines.deployment_pipeline import prediction_service_loader
from run_deployment import predominant

# Outline a worldwide variable to maintain monitor of the service standing
service_started = False

def start_service():
    international service_started
    service = prediction_service_loader(
        pipeline_name="continuous_deployment_pipeline",
        pipeline_step_name="mlflow_model_deployer_step",
        working=False,
    )
    service.begin(timeout=21)  # Begin the service
    service_started = True
    return service

def stop_service(service):
    international service_started
    service.cease()  # Cease the service
    service_started = False

def predominant():
    st.title("Worker Attrition Prediction")


    age = st.sidebar.slider("Age", 18, 65, 30)
    monthly_income = st.sidebar.slider("Month-to-month Revenue", 0, 20000, 5000)
    total_working_years = st.sidebar.slider("Whole Working Years", 0, 40, 10)
    years_in_current_role = st.sidebar.slider("Years in Present Position", 0, 20, 5)
    years_since_last_promotion = st.sidebar.slider("Years Since Final Promotion", 0, 15, 2)

    if st.button("Predict"):
        international service_started
        if not service_started:
            service = start_service()

        input_data = {
            "Age": [age],
            "MonthlyIncome": [monthly_income],
            "TotalWorkingYears": [total_working_years],
            "YearsInCurrentRole": [years_in_current_role],
            "YearsSinceLastPromotion": [years_since_last_promotion],
        }

        df = pd.DataFrame(input_data)
        json_list = json.masses(json.dumps(record(df.T.to_dict().values())))
        information = np.array(json_list)
        pred = service.predict(information)
        st.success(
            "Predicted Worker Attrition Likelihood (0 - 1): {:.2f}".format(
                pred[0]
            )
        )

        # Cease the service after prediction
        if service_started:
            stop_service(service)


if __name__ == "__main__":
    predominant()

Right here, we’ve got created a streamlit internet app, named “Worker Attrition Prediction“, during which customers can present the inputs resembling Age, month-to-month revenue and so forth., to make the prediction, when the person clicks the “Predict” button, the enter information is shipped to the deployed mannequin, the prediction is made and displayed for the person. This, is how our streamlit_app works. When, we run the streamlit_app.py file, we’ll get the community URL like this,

By clicking the community URL, we will see the superb Streamlit UI, used to make predictions.

You’ll be able to view all of your stacks, elements used, variety of pipelines ran within the ZenML Dashboard making your MLOps journey simple.

ZenML Dashboard:

Stacks:

Parts:

Variety of Pipelines:

Variety of runs:

Conclusion

Now we have efficiently constructed an Finish-to-Finish Worker Attrition Charge prediciton MLOps undertaking. Now we have ingested the information, cleaned it, educated the mannequin, consider the mannequin, set off the deployment, deploy the mannequin, predict the mannequin by getting the brand new information, seek for present mannequin companies, if current, then predict the information, get the person inputs from the Streamlit internet app and make predictions, whereas will assist the HR division to take information pushed selections.

GitHub Code: https://github.com/VishalKumar-S/Employee-attrition-rate-MLOps-Project

Key-takeaways

• ZenML acts as an highly effective orchestration device, with integration of different ML instruments.
• The Steady deployment pipeline makes positive, solely the most effective fashions are deployed, helps in predicting with excessive accuracy.
• Caching helps us in saving the assets and logging helps us monitor the pipeline, helps us in debugging and error monitoring.
• Dashboards assist us to have a transparent view on ML pipeline workflow.

Incessantly Requested Questions

The media proven on this article will not be owned by Analytics Vidhya and is used on the Writer’s discretion.

Associated