Getting Started

Let’s get started by downloading the prerequisites. ¹ To run the notebook, you will need a running Kubeflow cluster that includes a Minio service. Checkout the section for “Support Components” in Chapter 3 to configure Minio. Make sure that MinioClient (“mc”) is also installed.

We also need to prepare the data to facilitate training² and use Minio client to create the storage objects:

# Port-forward the Minio service to http://localhost:9000
kubectl port-forward -n kubeflow svc/minio-service 9000:9000 &

# Configure Minio host
mc config host add minio http://localhost:9000 minio minio123

# Create storage bucket
mc mb minio/data

# Copy storage objects
mc cp go/src/github.com/medium/items-recommender/data/recommend_1.csv \
        minio/data/recommender/users.csv
mc cp go/src/github.com/medium/items-recommender/data/trx_data.csv \
        minio/data/recommender/transactions.csv

Deploying a TensorFlow Training Job

So far we have done some TensorFlow training using Jupyter notebooks, which is a great way to prototype and experiment. But soon we may discover that our prototype is insufficient—perhaps we need to refine the model using more data, or perhaps we need to train the model using specialized hardware. Sometimes we may even need to continuously run the training job because our model is constantly evolving. Perhaps most importantly, our model has to be deployable to production, where it can serve actual customer requests.

In order to handle these requirements, our training code must be easily packageable and deployable to various different environments. One of the ways to achieve this is to use TFJob—a Kubernetes custom resource (implemented using Kubernetes operator tf-operator) that you can use to run TensorFlow training jobs on Kubernetes.

We’ll start by deploying our recommender as a single-container TFJob. Since we already have a Python notebook, exporting it as a Python file is fairly simple—just select “File,” then “Download as” and select “Python.” This should save your notebook as a ready-to-execute Python file.

The next step is to package the training code in a container. This can be done with the following Dockerfile.

FROM  tensorflow/tensorflow:1.15.2-py3
RUN pip3 install --upgrade pip
RUN pip3 install pandas --upgrade
RUN pip3 install keras --upgrade
RUN pip3 install minio --upgrade
RUN mkdir -p /opt/kubeflow
COPY Recommender_Kubeflow.py /opt/kubeflow/
ENTRYPOINT ["python3", "/opt/kubeflow/Recommender_Kubeflow.py"]

Next we need to build this container along with its required libraries, and push the container image to a repository.

docker build -t kubeflow/recommenderjob:1.0 .
docker push kubeflow/recommenderjob:1.0

Once that’s done, we are ready to create the specification for a TFJob.

apiVersion: "kubeflow.org/v1"   
kind: "TFJob"                   
metadata:
  name: "recommenderjob"        
spec:
  tfReplicaSpecs:               
    Worker:
      replicas: 1
    restartPolicy: Never
    template:
      spec:
        containers:
        - name: tensorflow image: kubeflow/recommenderjob:1.0

There are a few other optional configurations for your TFJob, including:

activeDeadlineSeconds

How long to keep this job active before the system can terminate it. If this is set, the system will kill the job after the deadline expires.

backoffLimit

How many times to keep retrying this job before marking it as failed. For example, setting this to 3 means that if a job fails 3 times, the system will stop retrying.

cleanPodPolicy

Configures whether or not to clean up the Kubernetes pods after the job completes. Setting this policy can be useful to keep pods for debugging purposes. This can be set to All (all pods are cleaned up), Running (only running pods are cleaned up), or None (no pods are cleaned up).

Deploy the TFJob to your cluster:

kubectl apply -f recommenderjob.yaml

You can monitor the status of the TFJob with the following command:

kubectl describe tfjob recommenderjob

This should display something like this:

Status:
  Completion Time:  2019-05-18T00:58:27Z
  Conditions:
    Last Transition Time:  2019-05-18T02:34:24Z
    Last Update Time:      2019-05-18T02:34:24Z
    Message:               TFJob recommenderjob is created.
    Reason:                TFJobCreated
    Status:                True
    Type:                  Created
    Last Transition Time:  2019-05-18T02:38:28Z
    Last Update Time:      2019-05-18T02:38:28Z
    Message:               TFJob recommenderjob is running.
    Reason:                TFJobRunning
    Status:                False
    Type:                  Running
    Last Transition Time:  2019-05-18T02:38:29Z
    Last Update Time:      2019-05-18T02:38:29Z
    Message:               TFJob recommenderjob successfully completed.
    Reason:                TFJobSucceeded
    Status:                True
    Type:                  Succeeded
  Replica Statuses:
    Worker:
      Succeeded:  1

Note that the status field contains a list of job conditions, which represent when the job transitioned into each state. This is useful for debugging—if the job failed, the reason for the job’s failure would appear here.

So far we have trained a fairly simple and straightforward model with a modest number of training samples. In real life, learning more complex models may require significantly more training samples or model parameters. Such models can be too large and computationally complex to be handled by one machine. This is where distributed training comes in.

Distributed Training

By now we’ve deployed a single-worker TensorFlow job with Kubeflow. It is called “single-worker” because everything from hosting the data to executing the actual training steps are done on a single machine. However, as models become more complex, a single machine is often insufficient—we may need to distribute the model or the training samples over several networked machines. TensorFlow supports a distributed training mode, in which training is performed in parallel over several worker nodes.

Distributed training typically comes in two flavors: data parallelism and model parallelism. In data parallelism, the training data is partitioned into chunks, and the same training code runs on each chunk. At the end of each training step, each worker communicates its updates to all other nodes. Model parallelism is the opposite—the same training data is used in all workers, but the model itself is partitioned. At the end of each step, each worker is responsible for synchronizing the shared parts of the model.

The TFJob interface supports multi-worker distributed training. Conceptually, a TFJob is a logical grouping of all resources related to a training job, including pods and services. In Kubeflow, each replicated worker or parameter server is scheduled on its own single-container pod. In order for the replicas to synchronize with each other, each replica needs to expose itself through an endpoint, which is a Kubernetes internal service. Grouping these resources logically under a parent resource (which is the TFJob) allows these resources to be co-scheduled and garbage collected together.

In this section we’ll deploy a simple MNist example with distributed training. The TensorFlow training code is provided for you at this GitHub repo.

Let’s take a look at the yaml file for the distributed TensorFlow job.

apiVersion: "kubeflow.org/v1"
kind: "TFJob"
metadata:
  name: "mnist"
  namespace: kubeflow
spec:
  cleanPodPolicy: None
  tfReplicaSpecs:
    Worker:
      replicas: 2
      restartPolicy: Never
      template:
        spec:
          containers:
            - name: tensorflow
              image: gcr.io/kubeflow-ci/tf-mnist-with-summaries:1.0
              command:
                - "python"
                - "/var/tf_mnist/mnist_with_summaries.py"
                - "--log_dir=/train/logs"
                - "--learning_rate=0.01"
                - "--batch_size=150"
              volumeMounts:
                - mountPath: "/train"
                  name: "training"
          volumes:
            - name: "training"
              persistentVolumeClaim:
                claimName: "tfevent-volume"

Note that the tfReplicaSpecs field now contains a few different replica types. In a typical TensorFlow training cluster, there are a few possible possibilities:

Chief

Responsible for orchestrating computational tasks, emitting events, and checkpointing the model.

Parameter Servers (PS)

Provide a distributed data store for the model parameters.

Worker

This is where the computations and training actually happen. When a Chief node is not explicitly defined (as in the above example), one of the workers acts as the Chief node.

Evaluator

The evaluators can be used to compute evaluation metrics as the model is trained.

A replica spec contains a number of properties that describe its desired state:

replicas

How many replicas should be spawned for this replica type.

template

A PodTemplateSpec that describes the pod to create for each replica.

restartPolicy

Determines whether pods will be restarted when they exit. The allowed values are as follows:

Always

Means the pod will always be restarted. This policy is good for parameter servers since they never exit and should always be restarted in the event of failure.

OnFailure

Means the pod will be restarted if the pod exits due to failure. A non-zero exit code indicates a failure. An exit code of 0 indicates success and the pod will not be restarted. This policy is good for chief and workers.

ExitCode

Means the restart behavior is dependent on the exit code of the tensorflow container as follows:

• 0 indicates the process completed successfully and will not be restarted.

• 1-127 indicates a permanent error and the container will not be restarted.

• 128-255 indicates a retryable error and the container will be restarted. This policy is good for the chief and workers.

Never

Means pods that terminate will never be restarted. This policy should rarely be used because Kubernetes will terminate pods for any number of reasons (e.g., node becomes unhealthy) and this will prevent the job from recovering.

Once you have the TFJob spec written, deploy it to your Kubeflow cluster:

kubectl apply -f dist-mnist.yaml

Monitoring the job status is similar to a single-container job:

kubectl describe tfjob mnist

This should output something like the following:

Status:
  Completion Time:  2019-05-12T00:58:27Z
  Conditions:
    Last Transition Time:  2019-05-12T00:57:31Z
    Last Update Time:      2019-05-12T00:57:31Z
    Message:               TFJob dist-mnist-example is created.
    Reason:                TFJobCreated
    Status:                True
    Type:                  Created
    Last Transition Time:  2019-05-12T00:58:21Z
    Last Update Time:      2019-05-12T00:58:21Z
    Message:               TFJob dist-mnist-example is running.
    Reason:                TFJobRunning
    Status:                False
    Type:                  Running
    Last Transition Time:  2019-05-12T00:58:27Z
    Last Update Time:      2019-05-12T00:58:27Z
    Message:               TFJob dist-mnist-example successfully completed.
    Reason:                TFJobSucceeded
    Status:                True
    Type:                  Succeeded
  Replica Statuses:
    Worker:
      Succeeded:  2

Notice that the Replica Statuses field shows a breakdown of status by each replica type. The TFJob is successfully completed when all of its workers complete. If any worker has failed, then the TFJob’s status would be failed as well.

Using GPUs

GPU (graphics processing units) are processors that are composed of many smaller and specialized cores. Originally designed to render graphics images, GPUs are increasingly used for massively parallel computational tasks, such as machine learning. Unlike CPUs, GPUs are ideal for distributing large workloads over its many cores and executing them concurrently.

To use GPUs for training, your Kubeflow cluster needs to be pre-configured to enable GPUs. Refer to the appendix section to see how to do this for your preferred cloud provider. After enabling GPUs on the cluster, you can enable GPUs on the specific replica type in the training spec by modifying the command-line arguments, for example:

    Worker:
      replicas: 4
      restartPolicy: Never
      template:
        spec:
          containers:
            - name: tensorflow
              image: kubeflow/tf-dist-mnist-test:1.0
              args:
            - python
            - /var/tf_dist_mnist/dist_mnist.py
            - --num_gpus=1

Using Other Frameworks for Distributed Training

Kubeflow is designed to be a multi-framework machine learning platform. That means the schema for distributed training can easily be extended to other frameworks. As of the time of this writing, there are a number of operators written to provide first-class support for other frameworks, including PyTorch and Caffe2.

As an example, here is what a PyTorch training job spec looks like ???.

apiVersion: "kubeflow.org/v1"
kind: "PyTorchJob"
metadata:
  name: "pytorch-dist"
spec:
  pytorchReplicaSpecs:
    Master:
      replicas: 1
      restartPolicy: OnFailure
      template:
        spec:
          containers:
            - name: pytorch
              image: gcr.io/kubeflow-ci/pytorch-dist-sendrecv-test:1.0
    Worker:
      replicas: 3
      restartPolicy: OnFailure
      template:
        spec:
          containers:
            - name: pytorch
              image: gcr.io/kubeflow-ci/pytorch-dist-sendrecv-test:1.0

As you can see, the format is very similar to that of TFJobs. The only difference is in the replica types.

Starting a New Notebook Session

Now let’s start by creating a new notebook. Similar to TensorFlow training, you can do this by navigating to the “Notebook Servers” panel in your Kubeflow dashboard, then clicking on “New Server” and following the on-screen instructions. For this example, we can use the tensorFlow-1.15.2-notebook-cpu:1.0 image.

When the notebook server starts up, click on the “Upload” button in the top right corner and upload the IncomePrediction.ipynb file. Click on the file to start a new session.

Data preparation

The first few sections of the notebook involves importing libraries and reading the data. Then we proceed to feature preparation.⁶ For feature transformation we are using Scikit-Learn pipelines. The pipeline makes it easier to feed the model with consistent data.

For our Random forest training we need to define ordinal (standardize data) and categorical (one-hot-encoding) features:

ordinal_features = [x for x in range(len(feature_names))
                if x not in list(category_map.keys())]
ordinal_transformer = Pipeline(steps=[
    ('imputer',  SimpleImputer(strategy='median')),
    ('scaler', StandardScaler())])


categorical_features = list(category_map.keys())
categorical_transformer = Pipeline(steps=[('imputer',
    SimpleImputer(strategy='median')),
    ('onehot', OneHotEncoder(handle_unknown='ignore'))])

Once features are defined, we can use a column transformer to combine them together:

preprocessor = ColumnTransformer(transformers=[
    ('num', ordinal_transformer, ordinal_features),
    ('cat', categorical_transformer, categorical_features)])
preprocessor.fit(X_train)

For our example, the transformer itself looks like the following:

ColumnTransformer(n_jobs=None, remainder='drop', sparse_threshold=0.3,
  transformer_weights=None,
  transformers=[('num',
    Pipeline(memory=None,
      steps=[
        ('imputer', SimpleImputer(add_indicator=False,
          copy=True,
          fill_value=None,
          missing_values=nan,
          strategy='median',
          verbose=0)),
        ('scaler', StandardScaler(copy=True,
          with_mean=True,
          with_std=True))],
        verbose=False),
      [0, 8, 9, 10]),
    ('cat',
     Pipeline(memory=None,
       steps=[('imputer', SimpleImputer(add_indicator=False,
         copy=True,
         fill_value=None,
         missing_values=nan,
         strategy='median',
         verbose=0)),
       ('onehot', OneHotEncoder(categories='auto',
         drop=None,
         dtype=<class 'numpy.float64'>,
         handle_unknown='ignore',
         sparse=True))],
       verbose=False),
       [1, 2, 3, 4, 5, 6, 7, 11])],
    verbose=False)

As a result of this transformation we have our data in the form of features ready for training.

Scikit-learn Training

Once we have our features prepared we can proceed with the training. Here we will use RandomForestClassifier provided by the Scikit-learn library:

np.random.seed(0)
clf = RandomForestClassifier(n_estimators=50)
clf.fit(preprocessor.transform(X_train), Y_train)

Once prediction is done, we can evaluate training results:

predict_fn = lambda x: clf.predict(preprocessor.transform(x))
print('Train accuracy: ', accuracy_score(Y_train, predict_fn(X_train)))
print('Test accuracy: ', accuracy_score(Y_test, predict_fn(X_test)))

Which returns the following results:

Train accuracy:  0.9655333333333334
Test accuracy:  0.855859375

At this point the model is created and can be directly used by exporting it (see below). One of most important attribute of a model is its explainability. Although model explainability is mostly used in model serving, it is also important for model creation, due to two main reasons:

• If explainability is important for model serving, during model creation, we often need to validate that the model that was created is explainable.

• Many of the model explanation methods require additional calculations during model creation.

Based on this, we will show here how to implement model explainability ⁷ during model creation.

Explaining the Model

For model explanation, we are using Anchors, part of Seldon’s Alibi project.

The algorithm provides model-agnostic (black box) and human interpretable explanations suitable for classification models applied to images, text and tabular data. The continuous features are discretized into quantiles (e.g., deciles), so they become more interpretable. The features in a candidate anchor are kept constant (same category or bin for discretized features) while we sample the other features from a training set.

explainer = AnchorTabular(
    predict_fn, feature_names, categorical_names=category_map, seed=1)
explainer.fit(X_train, disc_perc=[25, 50, 75])

This creates the following tabular Anchor:

AnchorTabular(meta={
    'name': 'AnchorTabular',
    'type': ['blackbox'],
    'explanations': ['local'],
    'params': {'seed': 1, 'disc_perc': [25, 50, 75]}
})

Now we can get an anchor for the prediction of the first observation in the test set. An anchor is a sufficient condition—that is, when the anchor holds, the prediction should be the same as the prediction for this instance:

idx = 0
class_names = adult.target_names
print('Prediction: ', class_names[explainer.predictor(X_test[idx].reshape(1, -1))[0]])

Which returns the following:

Prediction:  <=50K

We set the precision threshold to 0.95. This means that predictions on observations where the anchor holds will be the same as the prediction on the explained instance at least 95% of the time. Now we can get an explanation for this prediction:

explanation = explainer.explain(X_test[idx], threshold=0.95)
print('Anchor: %s' % (' AND '.join(explanation.anchor)))
print('Precision: %.2f' % explanation.precision)
print('Coverage: %.2f' % explanation.coverage)

Which returns:

Anchor: Marital Status = Separated AND Sex = Female
Precision: 0.95
Coverage: 0.18

Telling us that the main factors for decision are marital status (Separated) and sex (Female). Anchors might not be found for all points. Let’s try getting an anchor for a different observation in the test set—one for which the prediction is >50K:

idx = 6
class_names = adult.target_names
print('Prediction: ', class_names[explainer.predictor(X_test[idx].reshape(1, -1))[0]])

explanation = explainer.explain(X_test[idx], threshold=0.95)
print('Anchor: %s' % (' AND '.join(explanation.anchor)))
print('Precision: %.2f' % explanation.precision)
print('Coverage: %.2f' % explanation.coverage)

This will return:

Prediction:  >50K
Could not find an result satisfying the 0.95 precision constraint.
Now returning the best non-eligible result.
Anchor: Capital Loss > 0.00 AND Relationship = Husband AND
    Marital Status = Married AND Age > 37.00 AND
    Race = White AND Country = United-States AND Sex = Male
Precision: 0.71
Coverage: 0.05

Due to the imbalanced dataset (roughly 25:75 high:low earner proportion), during the sampling stage feature ranges corresponding to low-earners will be oversampled. As a result Anchor in this case is not found. This is a feature because it can point out an imbalanced dataset, but it can also be fixed by producing balanced datasets to enable anchors to be found for either class.

Exporting Model

In order to use the created model for serving, we need to export the model. This is done using Scikit-learn functionality:

dump(clf, '/tmp/job/income.joblib')

This exports a model in Scikit-learn format, that can be used by, for example, scikit-learn server for inference.

Integration into Pipelines

Regardless of which Python based machine learning library you want to use, if Kubeflow doesn’t have an operator for it you can simply write your code as normal and then containerize it. To take the notebook we built in this chapter and use it as a pipeline stage see “Using an Entire Notebook as a Dataprep Pipeline Stage”. Here we can use file_output to upload the resulting model to our artifact tracking system but you can also use the persistent volume mechanism.