Files
Ayush GuptaandGitHub 1c19d3a6eb Fix Istio sidecar injection by moving from annotations to labels (#2527)
* Fix Istio sidecar injection by moving from annotations to labels

Signed-off-by: madmecodes <[email protected]>

* Update Istio sidecar injection from annotations to labels across the codebase
Replace annotations with labels for Istio sidecar injection according to Istio recommendations. Update conformance tests, examples, constants, composers, and utilities to use the new label-based approach consistently.

Signed-off-by: madmecodes <[email protected]>

* fix: Update SuggestionLabels function and composer implementation for Istio label injection

Signed-off-by: madmecodes <[email protected]>

* Fix linting issues in mpi-job-horovod.py

Signed-off-by: madmecodes <[email protected]>

* update: function moved from annotations to labels

Signed-off-by: madmecodes <[email protected]>

---------

Signed-off-by: madmecodes <[email protected]>
2025-05-09 17:52:41 +00:00

219 KiB

Kubeflow Pipelines e2e mnist example

In this notebook you will create e2e mnist Kubeflow Pipeline to perfom:

  • Hyperparameter tuning using Katib
  • Distributive training with the best hyperparameters using TFJob
  • Serve the trained model using KFServing

Reference documentation:

Note: This Pipeline runs in the multi-user mode. Follow this guide to give your Notebook access to Kubeflow Pipelines.

In [1]:
# Install required packages (Kubeflow Pipelines and Katib SDK).
!pip install kfp==1.8.12
!pip install kubeflow-katib==0.13.0
Requirement already satisfied: kfp==1.8.12 in /opt/conda/lib/python3.8/site-packages (1.8.12)
Requirement already satisfied: tabulate<1,>=0.8.6 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (0.8.9)
Requirement already satisfied: google-cloud-storage<2,>=1.20.0 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (1.44.0)
Requirement already satisfied: google-api-core!=2.0.*,!=2.1.*,!=2.2.*,!=2.3.0,<3.0.0dev,>=1.31.5 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (2.7.1)
Requirement already satisfied: google-auth<2,>=1.6.1 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (1.35.0)
Requirement already satisfied: cloudpickle<3,>=2.0.0 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (2.0.0)
Requirement already satisfied: strip-hints<1,>=0.1.8 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (0.1.10)
Requirement already satisfied: typer<1.0,>=0.3.2 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (0.4.1)
Requirement already satisfied: requests-toolbelt<1,>=0.8.0 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (0.9.1)
Requirement already satisfied: protobuf<4,>=3.13.0 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (3.17.3)
Requirement already satisfied: absl-py<2,>=0.9 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (0.11.0)
Requirement already satisfied: typing-extensions<4,>=3.7.4 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (3.10.0.2)
Requirement already satisfied: uritemplate<4,>=3.0.1 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (3.0.1)
Requirement already satisfied: jsonschema<4,>=3.0.1 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (3.2.0)
Requirement already satisfied: Deprecated<2,>=1.2.7 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (1.2.13)
Requirement already satisfied: google-api-python-client<2,>=1.7.8 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (1.12.10)
Requirement already satisfied: click<9,>=7.1.2 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (7.1.2)
Requirement already satisfied: PyYAML<6,>=5.3 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (5.4.1)
Requirement already satisfied: fire<1,>=0.3.1 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (0.4.0)
Requirement already satisfied: kfp-server-api<2.0.0,>=1.1.2 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (1.6.0)
Requirement already satisfied: kubernetes<19,>=8.0.0 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (12.0.1)
Requirement already satisfied: kfp-pipeline-spec<0.2.0,>=0.1.14 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (0.1.14)
Requirement already satisfied: pydantic<2,>=1.8.2 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (1.9.0)
Requirement already satisfied: docstring-parser<1,>=0.7.3 in /opt/conda/lib/python3.8/site-packages (from kfp==1.8.12) (0.13)
Requirement already satisfied: six in /opt/conda/lib/python3.8/site-packages (from absl-py<2,>=0.9->kfp==1.8.12) (1.16.0)
Requirement already satisfied: wrapt<2,>=1.10 in /opt/conda/lib/python3.8/site-packages (from Deprecated<2,>=1.2.7->kfp==1.8.12) (1.13.3)
Requirement already satisfied: termcolor in /opt/conda/lib/python3.8/site-packages (from fire<1,>=0.3.1->kfp==1.8.12) (1.1.0)
Requirement already satisfied: googleapis-common-protos<2.0dev,>=1.52.0 in /opt/conda/lib/python3.8/site-packages (from google-api-core!=2.0.*,!=2.1.*,!=2.2.*,!=2.3.0,<3.0.0dev,>=1.31.5->kfp==1.8.12) (1.55.0)
Requirement already satisfied: requests<3.0.0dev,>=2.18.0 in /opt/conda/lib/python3.8/site-packages (from google-api-core!=2.0.*,!=2.1.*,!=2.2.*,!=2.3.0,<3.0.0dev,>=1.31.5->kfp==1.8.12) (2.25.1)
Requirement already satisfied: google-auth-httplib2>=0.0.3 in /opt/conda/lib/python3.8/site-packages (from google-api-python-client<2,>=1.7.8->kfp==1.8.12) (0.1.0)
Requirement already satisfied: httplib2<1dev,>=0.15.0 in /opt/conda/lib/python3.8/site-packages (from google-api-python-client<2,>=1.7.8->kfp==1.8.12) (0.20.4)
Requirement already satisfied: cachetools<5.0,>=2.0.0 in /opt/conda/lib/python3.8/site-packages (from google-auth<2,>=1.6.1->kfp==1.8.12) (4.2.4)
Requirement already satisfied: setuptools>=40.3.0 in /opt/conda/lib/python3.8/site-packages (from google-auth<2,>=1.6.1->kfp==1.8.12) (49.6.0.post20210108)
Requirement already satisfied: rsa<5,>=3.1.4 in /opt/conda/lib/python3.8/site-packages (from google-auth<2,>=1.6.1->kfp==1.8.12) (4.8)
Requirement already satisfied: pyasn1-modules>=0.2.1 in /opt/conda/lib/python3.8/site-packages (from google-auth<2,>=1.6.1->kfp==1.8.12) (0.2.8)
Requirement already satisfied: google-cloud-core<3.0dev,>=1.6.0 in /opt/conda/lib/python3.8/site-packages (from google-cloud-storage<2,>=1.20.0->kfp==1.8.12) (2.2.3)
Requirement already satisfied: google-resumable-media<3.0dev,>=1.3.0 in /opt/conda/lib/python3.8/site-packages (from google-cloud-storage<2,>=1.20.0->kfp==1.8.12) (2.3.2)
Requirement already satisfied: google-crc32c<2.0dev,>=1.0 in /opt/conda/lib/python3.8/site-packages (from google-resumable-media<3.0dev,>=1.3.0->google-cloud-storage<2,>=1.20.0->kfp==1.8.12) (1.3.0)
Requirement already satisfied: pyparsing!=3.0.0,!=3.0.1,!=3.0.2,!=3.0.3,<4,>=2.4.2 in /opt/conda/lib/python3.8/site-packages (from httplib2<1dev,>=0.15.0->google-api-python-client<2,>=1.7.8->kfp==1.8.12) (2.4.7)
Requirement already satisfied: pyrsistent>=0.14.0 in /opt/conda/lib/python3.8/site-packages (from jsonschema<4,>=3.0.1->kfp==1.8.12) (0.17.3)
Requirement already satisfied: attrs>=17.4.0 in /opt/conda/lib/python3.8/site-packages (from jsonschema<4,>=3.0.1->kfp==1.8.12) (21.2.0)
Requirement already satisfied: certifi in /opt/conda/lib/python3.8/site-packages (from kfp-server-api<2.0.0,>=1.1.2->kfp==1.8.12) (2021.5.30)
Requirement already satisfied: python-dateutil in /opt/conda/lib/python3.8/site-packages (from kfp-server-api<2.0.0,>=1.1.2->kfp==1.8.12) (2.8.1)
Requirement already satisfied: urllib3>=1.15 in /opt/conda/lib/python3.8/site-packages (from kfp-server-api<2.0.0,>=1.1.2->kfp==1.8.12) (1.26.5)
Requirement already satisfied: websocket-client!=0.40.0,!=0.41.*,!=0.42.*,>=0.32.0 in /opt/conda/lib/python3.8/site-packages (from kubernetes<19,>=8.0.0->kfp==1.8.12) (1.0.1)
Requirement already satisfied: requests-oauthlib in /opt/conda/lib/python3.8/site-packages (from kubernetes<19,>=8.0.0->kfp==1.8.12) (1.3.1)
Requirement already satisfied: pyasn1<0.5.0,>=0.4.6 in /opt/conda/lib/python3.8/site-packages (from pyasn1-modules>=0.2.1->google-auth<2,>=1.6.1->kfp==1.8.12) (0.4.8)
Requirement already satisfied: idna<3,>=2.5 in /opt/conda/lib/python3.8/site-packages (from requests<3.0.0dev,>=2.18.0->google-api-core!=2.0.*,!=2.1.*,!=2.2.*,!=2.3.0,<3.0.0dev,>=1.31.5->kfp==1.8.12) (2.10)
Requirement already satisfied: chardet<5,>=3.0.2 in /opt/conda/lib/python3.8/site-packages (from requests<3.0.0dev,>=2.18.0->google-api-core!=2.0.*,!=2.1.*,!=2.2.*,!=2.3.0,<3.0.0dev,>=1.31.5->kfp==1.8.12) (4.0.0)
Requirement already satisfied: wheel in /opt/conda/lib/python3.8/site-packages (from strip-hints<1,>=0.1.8->kfp==1.8.12) (0.36.2)
Requirement already satisfied: oauthlib>=3.0.0 in /opt/conda/lib/python3.8/site-packages (from requests-oauthlib->kubernetes<19,>=8.0.0->kfp==1.8.12) (3.2.0)
Requirement already satisfied: kubeflow-katib==0.13.0 in /opt/conda/lib/python3.8/site-packages (0.13.0)
Requirement already satisfied: setuptools>=21.0.0 in /opt/conda/lib/python3.8/site-packages (from kubeflow-katib==0.13.0) (49.6.0.post20210108)
Requirement already satisfied: certifi>=14.05.14 in /opt/conda/lib/python3.8/site-packages (from kubeflow-katib==0.13.0) (2021.5.30)
Requirement already satisfied: six>=1.10 in /opt/conda/lib/python3.8/site-packages (from kubeflow-katib==0.13.0) (1.16.0)
Requirement already satisfied: urllib3>=1.15.1 in /opt/conda/lib/python3.8/site-packages (from kubeflow-katib==0.13.0) (1.26.5)
Requirement already satisfied: kubernetes>=12.0.0 in /opt/conda/lib/python3.8/site-packages (from kubeflow-katib==0.13.0) (12.0.1)
Requirement already satisfied: pyyaml>=3.12 in /opt/conda/lib/python3.8/site-packages (from kubernetes>=12.0.0->kubeflow-katib==0.13.0) (5.4.1)
Requirement already satisfied: requests-oauthlib in /opt/conda/lib/python3.8/site-packages (from kubernetes>=12.0.0->kubeflow-katib==0.13.0) (1.3.1)
Requirement already satisfied: requests in /opt/conda/lib/python3.8/site-packages (from kubernetes>=12.0.0->kubeflow-katib==0.13.0) (2.25.1)
Requirement already satisfied: google-auth>=1.0.1 in /opt/conda/lib/python3.8/site-packages (from kubernetes>=12.0.0->kubeflow-katib==0.13.0) (1.35.0)
Requirement already satisfied: python-dateutil>=2.5.3 in /opt/conda/lib/python3.8/site-packages (from kubernetes>=12.0.0->kubeflow-katib==0.13.0) (2.8.1)
Requirement already satisfied: websocket-client!=0.40.0,!=0.41.*,!=0.42.*,>=0.32.0 in /opt/conda/lib/python3.8/site-packages (from kubernetes>=12.0.0->kubeflow-katib==0.13.0) (1.0.1)
Requirement already satisfied: pyasn1-modules>=0.2.1 in /opt/conda/lib/python3.8/site-packages (from google-auth>=1.0.1->kubernetes>=12.0.0->kubeflow-katib==0.13.0) (0.2.8)
Requirement already satisfied: cachetools<5.0,>=2.0.0 in /opt/conda/lib/python3.8/site-packages (from google-auth>=1.0.1->kubernetes>=12.0.0->kubeflow-katib==0.13.0) (4.2.4)
Requirement already satisfied: rsa<5,>=3.1.4 in /opt/conda/lib/python3.8/site-packages (from google-auth>=1.0.1->kubernetes>=12.0.0->kubeflow-katib==0.13.0) (4.8)
Requirement already satisfied: pyasn1<0.5.0,>=0.4.6 in /opt/conda/lib/python3.8/site-packages (from pyasn1-modules>=0.2.1->google-auth>=1.0.1->kubernetes>=12.0.0->kubeflow-katib==0.13.0) (0.4.8)
Requirement already satisfied: chardet<5,>=3.0.2 in /opt/conda/lib/python3.8/site-packages (from requests->kubernetes>=12.0.0->kubeflow-katib==0.13.0) (4.0.0)
Requirement already satisfied: idna<3,>=2.5 in /opt/conda/lib/python3.8/site-packages (from requests->kubernetes>=12.0.0->kubeflow-katib==0.13.0) (2.10)
Requirement already satisfied: oauthlib>=3.0.0 in /opt/conda/lib/python3.8/site-packages (from requests-oauthlib->kubernetes>=12.0.0->kubeflow-katib==0.13.0) (3.2.0)
In [2]:
import kfp
import kfp.dsl as dsl
from kfp import components

from kubeflow.katib import ApiClient
from kubeflow.katib import V1beta1ExperimentSpec
from kubeflow.katib import V1beta1AlgorithmSpec
from kubeflow.katib import V1beta1ObjectiveSpec
from kubeflow.katib import V1beta1ParameterSpec
from kubeflow.katib import V1beta1FeasibleSpace
from kubeflow.katib import V1beta1TrialTemplate
from kubeflow.katib import V1beta1TrialParameterSpec

Define the Pipelines tasks

To run this Pipeline, you should define:

  1. Katib hyperparameter tuning
  2. TFJob training
  3. KFServing inference

Step 1. Katib hyperparameter tuning task

Create the Kubeflow Pipelines task for the Katib hyperparameter tuning. This Experiment uses "random" algorithm and TFJob for the Trial's worker.

The Katib Experiment is similar to this example: https://github.com/kubeflow/katib/blob/master/examples/v1beta1/kubeflow-training-operator/tfjob-mnist-with-summaries.yaml.

In [3]:
# You should define the Experiment name, namespace and number of training steps in the arguments.
def create_katib_experiment_task(experiment_name, experiment_namespace, training_steps):
    # Trial count specification.
    max_trial_count = 5
    max_failed_trial_count = 3
    parallel_trial_count = 2

    # Objective specification.
    objective = V1beta1ObjectiveSpec(
        type="minimize",
        goal=0.001,
        objective_metric_name="loss"
    )

    # Algorithm specification.
    algorithm = V1beta1AlgorithmSpec(
        algorithm_name="random",
    )

    # Experiment search space.
    # In this example we tune learning rate and batch size.
    parameters = [
        V1beta1ParameterSpec(
            name="learning_rate",
            parameter_type="double",
            feasible_space=V1beta1FeasibleSpace(
                min="0.01",
                max="0.05"
            ),
        ),
        V1beta1ParameterSpec(
            name="batch_size",
            parameter_type="int",
            feasible_space=V1beta1FeasibleSpace(
                min="80",
                max="100"
            ),
        )
    ]

    # Experiment Trial template.
    # TODO (andreyvelich): Use community image for the mnist example.
    trial_spec = {
        "apiVersion": "kubeflow.org/v1",
        "kind": "TFJob",
        "spec": {
            "tfReplicaSpecs": {
                "Chief": {
                    "replicas": 1,
                    "restartPolicy": "OnFailure",
                    "template": {
                        "metadata": {
                            "labels": {
                                "sidecar.istio.io/inject": "false"
                            }
                        },
                        "spec": {
                            "containers": [
                                {
                                    "name": "tensorflow",
                                    "image": "docker.io/liuhougangxa/tf-estimator-mnist",
                                    "command": [
                                        "python",
                                        "/opt/model.py",
                                        "--tf-train-steps=" + str(training_steps),
                                        "--tf-learning-rate=${trialParameters.learningRate}",
                                        "--tf-batch-size=${trialParameters.batchSize}"
                                    ]
                                }
                            ]
                        }
                    }
                },
                "Worker": {
                    "replicas": 1,
                    "restartPolicy": "OnFailure",
                    "template": {
                        "metadata": {
                            "labels": {
                                "sidecar.istio.io/inject": "false"
                            }
                        },
                        "spec": {
                            "containers": [
                                {
                                    "name": "tensorflow",
                                    "image": "docker.io/liuhougangxa/tf-estimator-mnist",
                                    "command": [
                                        "python",
                                        "/opt/model.py",
                                        "--tf-train-steps=" + str(training_steps),
                                        "--tf-learning-rate=${trialParameters.learningRate}",
                                        "--tf-batch-size=${trialParameters.batchSize}"
                                    ]
                                }
                            ]
                        }
                    }
                }
            }
        }
    }

    # Configure parameters for the Trial template.
    trial_template = V1beta1TrialTemplate(
        primary_container_name="tensorflow",
        trial_parameters=[
            V1beta1TrialParameterSpec(
                name="learningRate",
                description="Learning rate for the training model",
                reference="learning_rate"
            ),
            V1beta1TrialParameterSpec(
                name="batchSize",
                description="Batch size for the model",
                reference="batch_size"
            ),
        ],
        trial_spec=trial_spec
    )

    # Create an Experiment from the above parameters.
    experiment_spec = V1beta1ExperimentSpec(
        max_trial_count=max_trial_count,
        max_failed_trial_count=max_failed_trial_count,
        parallel_trial_count=parallel_trial_count,
        objective=objective,
        algorithm=algorithm,
        parameters=parameters,
        trial_template=trial_template
    )

    # Create the KFP task for the Katib Experiment.
    # Experiment Spec should be serialized to a valid Kubernetes object.
    katib_experiment_launcher_op = components.load_component_from_url(
        "https://raw.githubusercontent.com/kubeflow/pipelines/master/components/kubeflow/katib-launcher/component.yaml")
    op = katib_experiment_launcher_op(
        experiment_name=experiment_name,
        experiment_namespace=experiment_namespace,
        experiment_spec=ApiClient().sanitize_for_serialization(experiment_spec),
        experiment_timeout_minutes=60,
        delete_finished_experiment=False)

    return op

Step 2. TFJob training task

Create the Kubeflow Pipelines task for the TFJob training. In this example TFJob runs the Chief and Worker with 1 replica.

Learn more about TFJob replica specifications in the Kubeflow docs: https://www.kubeflow.org/docs/components/training/tftraining/#what-is-tfjob.

In [4]:
# This function converts Katib Experiment HP results to args.
def convert_katib_results(katib_results) -> str:
    import json
    import pprint
    katib_results_json = json.loads(katib_results)
    print("Katib results:")
    pprint.pprint(katib_results_json)
    best_hps = []
    for pa in katib_results_json["currentOptimalTrial"]["parameterAssignments"]:
        if pa["name"] == "learning_rate":
            best_hps.append("--tf-learning-rate=" + pa["value"])
        elif pa["name"] == "batch_size":
            best_hps.append("--tf-batch-size=" + pa["value"])
    print("Best Hyperparameters: {}".format(best_hps))
    return " ".join(best_hps)
In [5]:
# You should define the TFJob name, namespace, number of training steps, output of Katib and model volume tasks in the arguments.
def create_tfjob_task(tfjob_name, tfjob_namespace, training_steps, katib_op, model_volume_op):
    import json
    # Get parameters from the Katib Experiment.
    # Parameters are in the format "--tf-learning-rate=0.01 --tf-batch-size=100"
    convert_katib_results_op = components.func_to_container_op(convert_katib_results)
    best_hp_op = convert_katib_results_op(katib_op.output)
    best_hps = str(best_hp_op.output)

    # Create the TFJob Chief and Worker specification with the best Hyperparameters.
    # TODO (andreyvelich): Use community image for the mnist example.
    tfjob_chief_spec = {
        "replicas": 1,
        "restartPolicy": "OnFailure",
        "template": {
            "metadata": {
                "labels": {
                    "sidecar.istio.io/inject": "false"
                }
            },
            "spec": {
                "containers": [
                    {
                        "name": "tensorflow",
                        "image": "docker.io/liuhougangxa/tf-estimator-mnist",
                        "command": [
                            "sh",
                            "-c"
                        ],
                        "args": [
                            "python /opt/model.py --tf-export-dir=/mnt/export --tf-train-steps={} {}".format(training_steps, best_hps)
                        ],
                        "volumeMounts": [
                            {
                                "mountPath": "/mnt/export",
                                "name": "model-volume"
                            }
                        ]
                    }
                ],
                "volumes": [
                    {
                        "name": "model-volume",
                        "persistentVolumeClaim": {
                            "claimName": str(model_volume_op.outputs["name"])
                        }
                    }
                ]
            }
        }
    }

    tfjob_worker_spec = {
        "replicas": 1,
        "restartPolicy": "OnFailure",
        "template": {
            "metadata": {
                "labels": {
                    "sidecar.istio.io/inject": "false"
                }
            },
            "spec": {
                "containers": [
                    {
                        "name": "tensorflow",
                        "image": "docker.io/liuhougangxa/tf-estimator-mnist",
                        "command": [
                            "sh",
                            "-c",
                        ],
                        "args": [
                          "python /opt/model.py --tf-export-dir=/mnt/export --tf-train-steps={} {}".format(training_steps, best_hps) 
                        ],
                    }
                ],
            }
        }
    }

    # Create the KFP task for the TFJob.
    tfjob_launcher_op = components.load_component_from_url(
        "https://raw.githubusercontent.com/kubeflow/pipelines/master/components/kubeflow/launcher/component.yaml")
    op = tfjob_launcher_op(
        name=tfjob_name,
        namespace=tfjob_namespace,
        chief_spec=json.dumps(tfjob_chief_spec),
        worker_spec=json.dumps(tfjob_worker_spec),
        tfjob_timeout_minutes=60,
        delete_finished_tfjob=False)
    return op

Step 3. KFServing inference

Create the Kubeflow Pipelines task for the KFServing inference.

In [6]:
# You should define the model name, namespace, output of the TFJob and model volume tasks in the arguments.
def create_kfserving_task(model_name, model_namespace, tfjob_op, model_volume_op):

    inference_service = '''
apiVersion: "serving.kubeflow.org/v1beta1"
kind: "InferenceService"
metadata:
  name: {}
  namespace: {}
  annotations:
    "sidecar.istio.io/inject": "false"
spec:
  predictor:
    tensorflow:
      storageUri: "pvc://{}/"
'''.format(model_name, model_namespace, str(model_volume_op.outputs["name"]))

    kfserving_launcher_op = components.load_component_from_url(
        'https://raw.githubusercontent.com/kubeflow/pipelines/master/components/kubeflow/kfserving/component.yaml')
    kfserving_launcher_op(action="create", inferenceservice_yaml=inference_service).after(tfjob_op)

Run the Kubeflow Pipeline

You should create the Kubeflow Pipeline from the above tasks.

In [7]:
name="mnist-e2e"
namespace="kubeflow-user-example-com"
training_steps="200"

@dsl.pipeline(
    name="End to End Pipeline",
    description="An end to end mnist example including hyperparameter tuning, train and inference"
)
def mnist_pipeline(name=name, namespace=namespace, training_steps=training_steps):
    # Run the hyperparameter tuning with Katib.
    katib_op = create_katib_experiment_task(name, namespace, training_steps)

    # Create volume to train and serve the model.
    model_volume_op = dsl.VolumeOp(
        name="model-volume",
        resource_name="model-volume",
        size="1Gi",
        modes=dsl.VOLUME_MODE_RWO
    )

    # Run the distributive training with TFJob.
    tfjob_op = create_tfjob_task(name, namespace, training_steps, katib_op, model_volume_op)

    # Create the KFServing inference.
    create_kfserving_task(name, namespace, tfjob_op, model_volume_op)

# Run the Kubeflow Pipeline in the user's namespace.
kfp_client=kfp.Client()
run_id = kfp_client.create_run_from_pipeline_func(mnist_pipeline, namespace=namespace, arguments={}).run_id
print("Run ID: ", run_id)
Run ID:  a6b4800d-6b45-43d0-979d-7c2e2dcc38aa

The finished Pipeline should look as follows.Kubeflow Pipeline.png

Predict from the trained model

Once Kubeflow Pipeline is finished, you are able to call the API endpoint with mnist image to predict from the trained model.

Note: If you are using Kubeflow + Dex setup and runing this Notebook outside of your Kubernetes cluster, follow this guide to get Session ID for the API requests.

In [8]:
import numpy as np
from PIL import Image
import requests

# Pipeline Run should be succeeded.
kfp_run = kfp_client.get_run(run_id=run_id)
if kfp_run.run.status == "Succeeded":
    print("Run {} has been Succeeded\n".format(run_id))

    # Specify the image URL here.
    image_url = "https://raw.githubusercontent.com/kubeflow/katib/master/examples/v1beta1/kubeflow-pipelines/images/9.bmp"
    image = Image.open(requests.get(image_url, stream=True).raw)
    data = np.array(image.convert('L').resize((28, 28))).astype(np.float).reshape(-1, 28, 28, 1)
    data_formatted = np.array2string(data, separator=",", formatter={"float": lambda x: "%.1f" % x})
    json_request = '{{ "instances" : {} }}'.format(data_formatted)

    # Specify the prediction URL. If you are runing this notebook outside of Kubernetes cluster, you should set the Cluster IP.
    url = "http://{}-predictor-default.{}.svc.cluster.local/v1/models/{}:predict".format(name, namespace, name)
    response = requests.post(url, data=json_request)

    print("Prediction for the image")
    display(image)
    print(response.json())
Run a6b4800d-6b45-43d0-979d-7c2e2dcc38aa has been Succeeded

Prediction for the image
<ipython-input-8-e6f2009e4aa7>:13: DeprecationWarning: `np.float` is a deprecated alias for the builtin `float`. To silence this warning, use `float` by itself. Doing this will not modify any behavior and is safe. If you specifically wanted the numpy scalar type, use `np.float64` here.
Deprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations
  data = np.array(image.convert('L').resize((28, 28))).astype(np.float).reshape(-1, 28, 28, 1)
{'predictions': [{'predictions': [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 1.0], 'classes': 9}]}
In [ ]: