End-to-End Machine Learning Solution with OpenText Vertica and Qwak Using VerticaPy

In our solution, we used 'Telco Customer Churn' dataset available in a CSV format. This dataset has 21 columns which includes,

• Churn - Customers that left within the last month.

• Services - Services of each customer (phone, multiple lines, internet, online security, online backup, device protection, tech support, and streaming TV and movies).

• Customer account information - How long they have been a customer, contract, payment method, paperless billing, monthly charges, and total charges.

• Customer demographics - Gender, age range, and if they have partners and dependents.

The goal is to see how these predictors influence churn and how you can detect customer churn using VerticaPy functionality.

If you want to explore this data, you can download the same file by registering on the Kaggle website, Churn Dataset.

In the following sections we will take you through the process of creating a model using Qwak to predict customer churn.

We loaded the sample dataset into Vertica with Jupyter Notebook using Verticapy .

1. Open Jupyter Notebook and install Verticapy by running the following command:

   pip install verticapy

2. Provide Vertica database connection details to connect to Vertica.

   import verticapy as vp
   # Creating a new connection
   vp.new_connection({"host": "<Verticahost>", 
                      "port": "5433", 
                      "database": "<Verticadbname>", 
                      "password": <password>, 
                      "user": "dbadmin"},
                      name = "MyVerticaConnection")
   # Connecting to the Database
   vp.connect("MyVerticaConnection")
   

3. Now, let’s create a new schema.

    vp.create_schema("qwak_schema")
   

4. Load Churn dataset into Vertica using the following command where we pass path of the downloaded dataset, table name, and schema name.

   vp.read_csv('Downloads/churn.csv', table_name = 'churn', schema = 'qwak_schema')

You can also use any ETL tool of your choice. For more information , see Importing Table Data for steps to load CSV data into a new or existing table.

1. Contact Qwak support to receive your API key and username.

2. Install Git and Install Python versions 3.7-3.9 using cmd. It is recommended to use a virtual environment.

3. Install Qwak Python SDK by running the following command from your console

   pip install --extra-index-url https://qwak:A3RX55aNSpE8dCV@qwak.jfrog.io/artifactory/api/pypi/qwak-pypi/simple "qwak-sdk<0.10"

4. After installing Qwak SDK, verify that the Qwak command is working

   

5. Authenticate your user and configure the Qwak environment to perform operations on Qwak platform by using Qwak configure command. Enter you API Key.

   

6. If authentication succeeds, you will get User successfully configured message.

   

You can create a project using Qwak UI or CLI commands:

Using Qwak UI

1. Login to Qwak cloud instance with your credentials.

   

2. Click Create new project.

   

3. Enter the project name and description. Click Create project.

   

Using CLI command

Run the following command to create a project:

qwak projects create --project-name "Vertica Workshop" --project-description "Vertica Workshop"

You can use any of the methods to create a project. You will see the project in the Qwak UI.

You can create a model using Qwak UI or CLI commands:

Using Qwak UI

1. Open the project and click Create new model.

   

2. Enter the model name and description. Click Create model.

   

Using CLI command

Run the following command to create a model using CLI command:

qwak models create --project-id {project_id} --model-name "churn_model"  --model-description "Predict churn customer"

You can use any method to create a model. You will see the model inside the project in the Qwak UI.

Generating Model Directory Structure

Qwak offers a default directory structure of a model.

You can generate the directory structure for your Qwak-based model using the following command:

qwak models init --model-directory churn_model --model-class-name ChurnClassifier  ~/

This will create a new directory named churn_model in your home directory. This contains main and tests directories with the required files such as init.py, conda.yml, model.py in the main directory and test_qwak_model.py in tests directory.

Main Directory

• The _init_.py file sets up the imports for Qwak model class, so that it is picked up by the model build process.

This file contains the following code:

from .model import ChurnClassifier

def load_model():
    return ChurnClassifier()

• The conda.yml is a dependency file that contains the external packages that the projects depend on. Qwak downloads and links the dependencies at build time.

Here, in this file we added VerticaPy to connect to the Vertica database.

name: ChurnClassifier
channels:
  - defaults
  - conda-forge
dependencies:
  - python=3.8
  - pip=22.2.2
  - pandas=1.1.5
  - pip:
      - --extra-index-url https://qwak:A3RX55aNSpE8dCV@qwak.jfrog.io/artifactory/api/pypi/qwak-pypi/simple qwak-sdk==0.9.134
      - verticapy==0.12.0

• The model.py creates a model class which defines two mandatory functions (build and predict) and two optional functions (schema and initialize_model).

The following code is the default code in the model.py file and you can update the code based on your requirement.

import pandas as pd
import qwak
from qwak.model.base import QwakModelInterface


class ChurnClassifier(QwakModelInterface):
    """ The Model class inherit QwakModelInterface base class
    """

    def __init__(self):
        pass

    def build(self):
        """ Responsible for loading the model. This method is invoked during build time (qwak build command)

           Example:
           >>> def build(self):
           >>>     ...
           >>>     train_pool = Pool(X_train, y_train, cat_features=categorical_features_indices)
           >>>     validate_pool = Pool(X_validation, y_validation, cat_features=categorical_features_indices)
           >>>     self.catboost.fit(train_pool, eval_set=validate_pool)
           """
        pass

    def schema(self):
        """ Specification of the model inputs and outputs. Optional method

        Example:
        >>> from qwak.model.schema import ModelSchema, Prediction, ExplicitFeature
        >>>
        >>> def schema(self) -> ModelSchema:
        >>>     model_schema = ModelSchema(
        >>>     features=[
        >>>         ExplicitFeature(name="State", type=str),
        >>>     ],
        >>>     predictions=[
        >>>         Prediction(name="score", type=float)
        >>>     ])
        >>>     return model_schema

       Returns: a model schema specification
       """
        pass

    def initialize_model(self):
        """
        Invoked when a model is loaded at serving time. Called once per model instance initialization. Can be used for
        loading and storing values that should only be available in a serving setting or loading pretrained models. Optional method

        Example usage:

        >>> def initialize_model(self):
        >>>     with open('model.pkl', 'rb') as infile:
        >>>         self.model = pickle.load(infile)
        """
        pass

    @qwak.api()
    def predict(self, df: pd.DataFrame) -> pd.DataFrame:
        """ Invoked on every API inference request.
        Args:
            pd (DataFrame): the inference vector, as a pandas dataframe

        Returns: model output (inference results), as a pandas dataframe
        """
        pass

Tests Directory

The test_qwak_model.py file defines integration tests that we run during the build process. The following code is the default code that you can update.

def test_model():
    """
        Skeleton for a test that will run as part of the build process
    """
    assert True 

You can update the code in the model.py file based on the requirement on how to connect Vertica database, prepare the data, build the model, and the inferencing. Here, we will show you two cases to update the code in model.py file based on the Inference done inside Qwak memory and inside Vertica database. Inference is the method of applying an ML model to a dataset and generating an output or “prediction”.

• Inferencing in Vertica Database

  The input data for inferencing can be in Vertica. We run the inference inside Vertica database by taking input data as a Pandas Dataframe with one column that has the table name which is already in Vertica to be used in prediction and then prepare the data for prediction. Run the predictions against the model where the model does the prediction inside Vertica. The prediction results are saved inside Vertica database.

  The following code explains the connection to the Vertica database, preparing the data, training the model, and prediction.

  import pandas as pd
  import verticapy as vp
  import qwak
  from verticapy.utilities import *
  from verticapy.learn.tools import *
  from qwak.model.base import QwakModelInterface
  from verticapy.learn.ensemble import RandomForestClassifier
  from verticapy.learn.preprocessing import OneHotEncoder
  
  # create connection with vertica DB
  CONN_INFO = {
              "host": "hostname",
              "port": "port",
              "database": "dbname",
              "password": "pwd",
              "user": "user"
          }
  vp.new_connection(CONN_INFO,name='qwak_conn')
  
  
  class ChurnClassifier(QwakModelInterface):
      """ The Model class inherit QwakModelInterface base class
      """
      def __init__(self):
          # Connect with the DB
          vp.connect('qwak_conn')
          # Initialize the model
          self.rf_model = RandomForestClassifier(name="qwak_schema.rf_churn",
                                                 n_estimators=10,
                                                 max_features="auto",
                                                 max_leaf_nodes=32,
                                                 sample=0.7,
                                                 max_depth=5,
                                                 min_samples_leaf=1,
                                                 min_info_gain=0.0,
                                                 nbins=32)
  
      @staticmethod
      def data_preparation(vdf, train = True):
          """method to prepare data"""
  
          vdf["TotalCharges"].dropna()
          for column in [
              "DeviceProtection",
              "MultipleLines",
              "PaperlessBilling",
              "TechSupport",
              "Partner",
              "StreamingTV",
              "OnlineBackup",
              "Dependents",
              "OnlineSecurity",
              "PhoneService",
              "StreamingMovies",
              ]:
              vdf[column].decode("Yes", 1, 0)
          if train:
              enc = OneHotEncoder(name = "qwak_schema.ohe_churn")
  
              enc.fit(vdf,
                      ["gender", "Contract", "PaymentMethod", "InternetService"])
  
              vdf = enc.transform(vdf,
                                      ["gender", "Contract", "PaymentMethod", "InternetService"])
          else:
              model_enc = load_model("qwak_schema.ohe_churn")
              vdf = model_enc.transform(vdf,
                                      ["gender", "Contract", "PaymentMethod", "InternetService"])
              
          vdf.drop(["gender", "Contract", "PaymentMethod", "InternetService"])
          return vdf
  
      def build(self):
          """ 
          Responsible for loading the model. This method is invoked during build time (qwak build command)
          This function train and save the model inside vertica using data inside vertica
          """
          # Read the data from DB and prepare it for model training
          churn = ChurnClassifier.data_preparation(vp.vDataFrame("qwak_schema.churn"))
          churn['churn'].decode("Yes", 1, 0)
          # Split the data to train and test
          train, test = churn.train_test_split(test_size=0.2, random_state=0)
  
          # drop the model if already exist
          self.rf_model.drop()
          # features
          predictors = train.get_columns(exclude_columns=['churn', 'customerID'])
          # fit the model
          self.rf_model.fit(input_relation=train,
                            X=predictors,
                            y='churn',
                            test_relation=test)
          # Compute the metric score
          accuracy_score = self.rf_model.score("accuracy")
          qwak.log_metric({"accuracy_score": accuracy_score})
          
  
      @qwak.api()
      def predict(self, input_data: pd.DataFrame) -> pd.DataFrame:
          """ Invoked on every API inference request.
          Args:
              input_data (Dataframe): dataframe with one column that have the table name to use for prediction.
  
          Returns: model output (inference results), as dataframe with the size of the input, f1 and accuracy metrics to make sure that the prediction is preformed.
          """
          # get the table name from the input
          table_name = input_data['table_name'].values[0]
          # read the table
          input_data = vp.vDataFrame(table_name, schema="qwak_schema")
          # prepare the data for prediction
          input_data = ChurnClassifier.data_preparation(input_data, train=False)
          input_data['churn'].decode("Yes", 1, 0)
          # list of the independent variables
          predictors = input_data.get_columns(exclude_columns=['churn', 'customerID'])
          # predict
          output_data = self.rf_model.predict(input_data, predictors, name='churn_pred')
          # save the results in DB
          output_data.to_db(name='"qwak_schema"."churn_ref"', relation_type="insert")
          # return the output to pandas DataFrame
          return output_data.to_pandas()
  
  
  

• Inferencing in Qwak Memory

  The input data for inferencing can be in memory. We run the inference on memory by taking pandas dataframe as input and converting the Vertica model to memmodel and then the model does the prediction inside Qwak memory.

  It returns Pandas Dataframe as output and we are saving the results in Vertica database by converting pandas to Vertica dataframe.

import pandas as pd
import verticapy as vp
import qwak
from verticapy.utilities import *
from qwak.model.base import QwakModelInterface
from verticapy.learn.ensemble import RandomForestClassifier
from verticapy.learn.preprocessing import OneHotEncoder
from verticapy.learn.tools import *
import warnings
warnings.filterwarnings('ignore')

CONN_INFO = {
            "host": "hostname",
            "port": "port",
            "database": "dbname",
            "password": "pwd",
            "user": "user"
        }
vp.new_connection(CONN_INFO,name='qwak_conn')



class churnClassifier(QwakModelInterface):
    """The Model class inherit QwakModelInterface base class

    In this case we train and score using in memory model
    """

    def __init__(self):
        # Connect with the DB
        vp.connect('qwak_conn')
        # Initialize the model
        self.rf_model = RandomForestClassifier(
            name="qwak_schema.rf_churn",
            n_estimators=10,
            max_features="auto",
            max_leaf_nodes=32,
            sample=0.7,
            max_depth=5,
            min_samples_leaf=1,
            min_info_gain=0.0,
            nbins=32,
        )

    @staticmethod
    def data_preparation(vdf, train = True):
        """method to prepare data"""

        vdf["TotalCharges"].dropna()
        for column in [
            "DeviceProtection",
            "MultipleLines",
            "PaperlessBilling",
            "TechSupport",
            "Partner",
            "StreamingTV",
            "OnlineBackup",
            "Dependents",
            "OnlineSecurity",
            "PhoneService",
            "StreamingMovies",
            ]:
            vdf[column].decode("Yes", 1, 0)
        if train:
            enc = OneHotEncoder(name = "qwak_schema.ohe_churn")

            enc.fit(vdf,
                    ["gender", "Contract", "PaymentMethod", "InternetService"])

            vdf = enc.transform(vdf,
                                 ["gender", "Contract", "PaymentMethod", "InternetService"])
        else:
            model_enc = load_model("qwak_schema.ohe_churn")
            vdf = model_enc.transform(vdf,
                                 ["gender", "Contract", "PaymentMethod", "InternetService"])
            
        vdf.drop(["gender", "Contract", "PaymentMethod", "InternetService"])
        return vdf

    def build(self):
        """Responsible for loading the model. This method is invoked during build time (qwak build command)"""
        # Read the data from DB and prepare it for model training
        churn = churnClassifier.data_preparation(vp.vDataFrame("qwak_schema.churn"))
        churn['churn'].decode("Yes", 1, 0)
        # Split the data to train and test
        train, test = churn.train_test_split(test_size=0.2, random_state=0)

        # drop the model if already exist
        self.rf_model.drop()
        predictors = churn.get_columns(exclude_columns=["churn", "CustomerID"])
        # fit the model
        self.rf_model.fit(input_relation=train, 
            X=predictors, 
            y="churn", 
            test_relation=test
        )
        # Compute the metric score
        accuracy_score = self.rf_model.score("accuracy")
        qwak.log_metric({"accuracy_score": accuracy_score})


    @qwak.api()
    def predict(self, input_data: pd.DataFrame) -> pd.DataFrame:
        """
        Invoked on every API inference request.
        This method take DataFrame as input, then the model does the
        prediction. At the end the results are saved in Vetica and return DataFrame

        Args:
            input_data (list/DataFrame): the inference vector, as a list of lists

        Returns: model output (inference results), as a DataFrame
        """
         # convert the model to memmodel
        m_model = self.rf_model.to_memmodel()
        # selected features 
        predictors = ['gender_1', 'seniorcitizen','partner','dependents','tenure', 'phoneservice','multiplelines',
                        'InternetService_1','InternetService_2','onlinesecurity','onlinebackup','deviceprotection','techsupport',
                        'streamingtv','streamingmovies','Contract_1','Contract_2','paperlessbilling','PaymentMethod_1',
                        'PaymentMethod_2','PaymentMethod_3', 'monthlycharges','totalcharges']
        # create pandas df of the predictions
        input_data['churn_pred'] = m_model.predict(input_data[predictors])
        # save dat in vertica
        pandas_to_vertica(df = input_data, name = "churn_ref", schema = "qwak_schema", insert=True)
        
        return input_data

You can update the code in test_qwak_model.py as shown in the following:

• Inferencing in Vertica Database

  The input data given for integration test is a table that is already stored in Vertica.

  import pandas as pd
  from qwak_mock import real_time_client
  
  
  def test_churn_classifier(real_time_client):
  
      feature_vector = [{"table_name":"churn_to_pred_1"}]
  
      predict_res = real_time_client.predict(feature_vector)
      assert predict_res['churn_pred'].values[0] == '0'
  

• Inferencing in Qwak Memory

The input data given for integration test is already prepared data outside Qwak.

from qwak_mock import real_time_client

def test_churn_classifier(real_time_client):
    feature_vector = [{'customerid': '0137-OCGAB',
            'gender_1': 0,
            'seniorcitizen': 0,
            'partner': 0,
            'dependents': 0,
            'tenure': 1,
            'phoneservice': 1,
            'multiplelines': 1,
            'InternetService_1': 1,
            'InternetService_2': 0,
            'onlinesecurity': 0,
            'onlinebackup': 1,
            'deviceprotection': 0,
            'techsupport': 0,
            'streamingtv': 0,
            'streamingmovies': 0,
            'Contract_1': 0,
            'Contract_2': 0,
            'paperlessbilling': 1,
            'PaymentMethod_1': 0,
            'PaymentMethod_2': 0,
            'PaymentMethod_3': 1,
            'monthlycharges': 80.200,
            'totalcharges': 80.200,
            'churn': True} ]

    churn_pred = real_time_client.predict(feature_vector)
    assert churn_pred['churn_pred'].values[0]==1

After you are done with updating the files in model folder, change the directory to model folder in CLI and run the following command to run the model build process:

qwak models build --model-id "churn_model" .

The build process may take several minutes. You can view the builds created inside the model after you the run the model.

You can view the logs for the various builds in the build page from the Logs tab by clicking the build id.

You can view the code for various builds in the build page from the Code tab.

After a successful build, deploy the model as an endpoint running in the Qwak platform.

To deploy a model from the Qwak UI:

1. Click Deploy in the builds page of the model.

   
   

2. Select the deployment method and click Next. Here we selected Realtime.

   

3. Configure your real-time deployment by selecting the initial number of pods, CPU fraction, and memory.

   

   Configure advanced settings such as injecting environment variables, setting invocation timeouts, and more and then click Deploy.

   

4. You will see that the build is deployed in the build page.

   

Automating model build and deployment helps maintain model accuracy in production. It keeps your model accurate by automatically retraining and deploying based on a cron expression, defined time interval, or metric base triggers.

1. In Automations page of the model, click Create your first automation.

   

2. The automated training is triggered by either the time interval or a condition. You can automate by time schedule, even daily, hourly, weekly or you can use CRON for another period. Click Continue.

   

3. You need to push your code to Git so that the platform pulls code when required and then provide the Git URL, branch, and resources.

   

4. Scroll down and set the deployment specification.

   

5. You can then set deployment condition based on a metric that you prefer. You can use a threshold for that, so whenever the new model is trained, it verifies if its score met the threshold condition and will be automatically deployed. Click Continue.

   

6. Provide a name for the automation and click Save.

   

7. Click Run now. It runs as another version of the model. You can check the resources allocated for this automation and edit it when required.

   

8. In the Builds section you will see your version that has been created using the automation and this will be automatically deployed based on the deployment condition.

   

After deploying the model, run predictions against the model by executing a python file where we give customer details as input data and run the prediction, you will see the result of the prediction whether the customer is likely to churn or not.

• Inferencing in Vertica Database

  If you have deployed the build to run the prediction inside Vertica database, then execute the following python file.

  

  from qwak_inference import RealTimeClient
  
  model_id = "churn_model"
  environment_name = "vertica"
  
  feature_vector = [{"table_name":"churn_to_pred_1"}]
  
  client = RealTimeClient(model_id=model_id, environment=environment_name)
  response = client.predict(feature_vector)
  print(response)
  
  

• Inferencing in Qwak Memory

  If you have deployed the build to run the prediction in Qwak memory, then execute the following python file.

  

  from qwak_inference import RealTimeClient
  
  model_id = "churn_model"
  environment_name = "vertica"
  
  feature_vector = [{'customerid': '0137-OCGAB', 
                     'gender_1': 0, 
                     'seniorcitizen': 0, 
                     'partner': 0, 
                     'dependents': 0, 
                     'tenure': 1, 
                     'phoneservice': 1, 
                     'multiplelines': 1, 
                     'InternetService_1': 1, 
                     'InternetService_2': 0, 
                     'onlinesecurity': 0, 
                     'onlinebackup': 1, 
                     'deviceprotection': 0, 
                     'techsupport': 0, 
                     'streamingtv': 0, 
                     'streamingmovies': 0, 
                     'Contract_1': 0, 
                     'Contract_2': 0, 
                     'paperlessbilling': 1, 
                     'PaymentMethod_1': 0, 
                     'PaymentMethod_2': 0, 
                     'PaymentMethod_3': 1, 
                     'monthlycharges': 80.200, 
                     'totalcharges': 80.200, 
                     'churn': True}]
  
  client = RealTimeClient(model_id=model_id, environment=environment_name)
  response = client.predict(feature_vector)
  print(response)
  
  

You can view the results of prediction from Jupyter Notebook that we have already saved in Vertica database while performing prediction either inside Vertica database or in Qwak memory.

Once you start running predictions against the model, you will start seeing the relevant metrics in the Health dashboard in the Overview tab.

You can view every prediction that you have invoked against the model by querying the model in Analytics section.

1. Click Analytics from the model page.

2. Enter the query as per your requirement.

3. Click Run. You will see a table containing a row for every prediction invoked against the model.