<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Kubeflow E2E MNIST Case
Step1:  Configure the Docker Registry for Kubeflow Fairing
In order to build docker images from your notebook we need a docker registry where the images will be stored
Note
Step2:  Create PV/PVC to Store the Exported Model
Create Persistent Volume(PV) and Persistent Volume Claim(PVC), the PVC will be used by pods of training and serving for local mode in steps below.
Note
Step3:  (Optional) Skip below creating PV/PVC step if you set an existing PV and PVC.
Step4:  Use Kubeflow fairing to build the docker image and launch a TFJob for training
Use kubeflow fairing to build a docker image that includes all your dependencies
Launch a TFJob in the on premise cluster to taining model.
Firstly set some custom training parameters for TFJob.
Step5:  Use Kubeflow Fairing to build a docker image and push to docker registry, and then launch a TFJob in the on-prem cluster for distributed training model.
Step6:  Get the Created TFJobs
Step7:  Wait For the Training Job to finish
Step8:  Check if the TFJob succeeded.
Step9:  Get the Training Logs
Step10:  Deploy Service using KFServing
Step11:  Get the InferenceService
Step12:  Get the InferenceService and Service Endpoint
Step13:  Run a prediction to the InferenceService
Step14:  Clean Up
Delete the TFJob
Step15:  Delete the InferenceService.
<ASSISTANT_TASK:>
Python Code:
!pip show kubeflow-fairing
Explanation: Kubeflow E2E MNIST Case: Building, Distributed Training and Serving
This example guides you through:
  1. Taking an example TensorFlow model and modifying it to support distributed training.
  1. Using Kubeflow Fairing to build docker image and launch a TFJob to train model.
  1. Using Kubeflow Fairing to create InferenceService (KFServing) for the trained model.
  1. Clean up the TFJob and InferenceService using kubeflow-tfjob and kfserving SDK client.
Requirements
The Kubeflow Fairing, TF-Operator and KFServing have been installed in Kubenertes Cluster.
Prepare Training Code
We modified the examples to be better suited for distributed training and model serving. There is a delta between existing distributed mnist examples and what's needed to run well as a TFJob. The updated training code is mnist.py. 
Install Required Libraries
End of explanation
# Set docker registry to store image.
# Ensure you have permission for pushing docker image requests. 
DOCKER_REGISTRY = 'index.docker.io/jinchi'
# Set namespace. Note that the created PVC should be in the namespace.
my_namespace = 'hejinchi'
# You also can get the default target namepspace using below API.
#namespace = fairing_utils.get_default_target_namespace()
Explanation: Configure the Docker Registry for Kubeflow Fairing
In order to build docker images from your notebook we need a docker registry where the images will be stored
Note: The below section must be updated to your values.
End of explanation
# To satify the distributed training, the PVC should be access from all nodes in the cluster.
# The example creates a NFS PV to satify that.
nfs_server = '172.16.189.69'
nfs_path = '/opt/kubeflow/data/mnist'
pv_name = 'kubeflow-mnist'
pvc_name = 'mnist-pvc'
Explanation: Create PV/PVC to Store the Exported Model
Create Persistent Volume(PV) and Persistent Volume Claim(PVC), the PVC will be used by pods of training and serving for local mode in steps below.
Note: The below section must be updated to your values.
End of explanation
from kubernetes import client as k8s_client
from kubernetes import config as k8s_config
from kubeflow.fairing.utils import is_running_in_k8s
pv_yaml = f'''
apiVersion: v1
kind: PersistentVolume
metadata:
  name: {pv_name}
spec:
  capacity:
    storage: 10Gi
  accessModes:
  - ReadWriteMany
  persistentVolumeReclaimPolicy: Retain
  nfs:
    path: {nfs_path}
    server: {nfs_server}
'''
pvc_yaml = f'''
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: {pvc_name}
  namespace: {my_namespace}
spec:
  accessModes:
    - ReadWriteMany
  storageClassName: ""
  resources:
    requests:
      storage: 10Gi
'''
if is_running_in_k8s():
    k8s_config.load_incluster_config()
else:
    k8s_config.load_kube_config()
k8s_core_api = k8s_client.CoreV1Api()
k8s_core_api.create_persistent_volume(yaml.safe_load(pv_yaml))
k8s_core_api.create_namespaced_persistent_volume_claim(my_namespace, yaml.safe_load(pvc_yaml))
Explanation: (Optional) Skip below creating PV/PVC step if you set an existing PV and PVC.
End of explanation
num_chief = 1 #number of Chief in TFJob 
num_ps = 1  #number of PS in TFJob 
num_workers = 2  #number of Worker in TFJob 
model_dir = "/mnt"
export_path = "/mnt/export" 
train_steps = "1000"
batch_size = "100"
learning_rate = "0.01"
Explanation: Use Kubeflow fairing to build the docker image and launch a TFJob for training
Use kubeflow fairing to build a docker image that includes all your dependencies
Launch a TFJob in the on premise cluster to taining model.
Firstly set some custom training parameters for TFJob.
End of explanation
import uuid
from kubeflow import fairing   
from kubeflow.fairing.kubernetes.utils import mounting_pvc
tfjob_name = f'mnist-training-{uuid.uuid4().hex[:4]}'
output_map =  {
    "Dockerfile": "Dockerfile",
    "mnist.py": "mnist.py"
}
command=["python",
         "/opt/mnist.py",
         "--tf-model-dir=" + model_dir,
         "--tf-export-dir=" + export_path,
         "--tf-train-steps=" + train_steps,
         "--tf-batch-size=" + batch_size,
         "--tf-learning-rate=" + learning_rate]
fairing.config.set_preprocessor('python', command=command, path_prefix="/app", output_map=output_map)
fairing.config.set_builder(name='docker', registry=DOCKER_REGISTRY, base_image="",
                           image_name="mnist", dockerfile_path="Dockerfile")
fairing.config.set_deployer(name='tfjob', namespace=my_namespace, stream_log=False, job_name=tfjob_name,
                            chief_count=num_chief, worker_count=num_workers, ps_count=num_ps, 
                            pod_spec_mutators=[mounting_pvc(pvc_name=pvc_name, pvc_mount_path=model_dir)])
fairing.config.run()
Explanation: Use Kubeflow Fairing to build a docker image and push to docker registry, and then launch a TFJob in the on-prem cluster for distributed training model.
End of explanation
from kubeflow.tfjob import TFJobClient
tfjob_client = TFJobClient()
tfjob_client.get(tfjob_name, namespace=my_namespace)
Explanation: Get the Created TFJobs
End of explanation
tfjob_client.wait_for_job(tfjob_name, namespace=my_namespace, watch=True)
Explanation: Wait For the Training Job to finish
End of explanation
tfjob_client.is_job_succeeded(tfjob_name, namespace=my_namespace)
Explanation: Check if the TFJob succeeded.
End of explanation
tfjob_client.get_logs(tfjob_name, namespace=my_namespace)
Explanation: Get the Training Logs
End of explanation
from kubeflow.fairing.deployers.kfserving.kfserving import KFServing
isvc_name = f'mnist-service-{uuid.uuid4().hex[:4]}'
isvc = KFServing('tensorflow', namespace=my_namespace, isvc_name=isvc_name,
                 default_storage_uri='pvc://' + pvc_name + '/export')
isvc.deploy(isvc.generate_isvc())
Explanation: Deploy Service using KFServing
End of explanation
from kfserving import KFServingClient
kfserving_client = KFServingClient()
kfserving_client.get(namespace=my_namespace)
Explanation: Get the InferenceService
End of explanation
mnist_isvc = kfserving_client.get(isvc_name, namespace=my_namespace)
mnist_isvc_name = mnist_isvc['metadata']['name']
mnist_isvc_endpoint = mnist_isvc['status'].get('url', '')
print("MNIST Service Endpoint: " + mnist_isvc_endpoint)
Explanation: Get the InferenceService and Service Endpoint
End of explanation
ISTIO_CLUSTER_IP=!kubectl -n istio-system get service istio-ingressgateway -o jsonpath='{.spec.clusterIP}'
CLUSTER_IP=ISTIO_CLUSTER_IP[0]
MODEL_HOST=f"Host: {mnist_isvc_name}.{my_namespace}.example.com"
!curl -v -H "{MODEL_HOST}" http://{CLUSTER_IP}/v1/models/{mnist_isvc_name}:predict -d @./input.json
Explanation: Run a prediction to the InferenceService
End of explanation
tfjob_client.delete(tfjob_name, namespace=my_namespace)
Explanation: Clean Up
Delete the TFJob
End of explanation
kfserving_client.delete(isvc_name, namespace=my_namespace)
Explanation: Delete the InferenceService.
End of explanation
<END_TASK>