<jupyter_start><jupyter_text>Continuous training pipeline with KFP and Cloud AI Platform **Learning Objectives:**1. Learn how to use KF pre-build components (BiqQuery, CAIP training and predictions)1. Learn how to use KF lightweight python components1. Learn how to build a KF pipeline with these components1. Learn how to compile, upload, and run a KF pipeline with the command lineIn this lab, you will build, deploy, and run a KFP pipeline that orchestrates **BigQuery** and **Cloud AI Platform** services to train, tune, and deploy a **scikit-learn** model.  Understanding the pipeline design The workflow implemented by the pipeline is defined using a Python based Domain Specific Language (DSL). The pipeline's DSL is in the `covertype_training_pipeline.py` file that we will generate below.The pipeline's DSL has been designed to avoid hardcoding any environment specific settings like file paths or connection strings. These settings are provided to the pipeline code through a set of environment variables.<jupyter_code>!grep 'BASE_IMAGE =' -A 5 pipeline/covertype_training_pipeline.py<jupyter_output>BASE_IMAGE = os.getenv('BASE_IMAGE')
TRAINER_IMAGE = os.getenv('TRAINER_IMAGE')
RUNTIME_VERSION = os.getenv('RUNTIME_VERSION')
PYTHON_VERSION = os.getenv('PYTHON_VERSION')
COMPONENT_URL_SEARCH_PREFIX = os.getenv('COMPONENT_URL_SEARCH_PREFIX')
USE_KFP_SA = os.getenv('USE_KFP_SA')<jupyter_text>The pipeline uses a mix of custom and pre-build components.- Pre-build components. The pipeline uses the following pre-build components that are included with the KFP distribution:    - [BigQuery query component](https://github.com/kubeflow/pipelines/tree/0.2.5/components/gcp/bigquery/query)    - [AI Platform Training component](https://github.com/kubeflow/pipelines/tree/0.2.5/components/gcp/ml_engine/train)    - [AI Platform Deploy component](https://github.com/kubeflow/pipelines/tree/0.2.5/components/gcp/ml_engine/deploy)- Custom components. The pipeline uses two custom helper components that encapsulate functionality not available in any of the pre-build components. The components are implemented using the KFP SDK's [Lightweight Python Components](https://www.kubeflow.org/docs/pipelines/sdk/lightweight-python-components/) mechanism. The code for the components is in the `helper_components.py` file:    - **Retrieve Best Run**. This component retrieves a tuning metric and hyperparameter values for the best run of a AI Platform Training hyperparameter tuning job.    - **Evaluate Model**. This component evaluates a *sklearn* trained model using a provided metric and a testing dataset.<jupyter_code>%%writefile ./pipeline/covertype_training_pipeline.py
# Copyright 2019 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""KFP pipeline orchestrating BigQuery and Cloud AI Platform services."""

import os

from helper_components import evaluate_model
from helper_components import retrieve_best_run
from jinja2 import Template
import kfp
from kfp.components import func_to_container_op
from kfp.dsl.types import Dict
from kfp.dsl.types import GCPProjectID
from kfp.dsl.types import GCPRegion
from kfp.dsl.types import GCSPath
from kfp.dsl.types import String
from kfp.gcp import use_gcp_secret

# Defaults and environment settings
BASE_IMAGE = os.getenv('BASE_IMAGE')
TRAINER_IMAGE = os.getenv('TRAINER_IMAGE')
RUNTIME_VERSION = os.getenv('RUNTIME_VERSION')
PYTHON_VERSION = os.getenv('PYTHON_VERSION')
COMPONENT_URL_SEARCH_PREFIX = os.getenv('COMPONENT_URL_SEARCH_PREFIX')
USE_KFP_SA = os.getenv('USE_KFP_SA')

TRAINING_FILE_PATH = 'datasets/training/data.csv'
VALIDATION_FILE_PATH = 'datasets/validation/data.csv'
TESTING_FILE_PATH = 'datasets/testing/data.csv'

# Parameter defaults
SPLITS_DATASET_ID = 'splits'
HYPERTUNE_SETTINGS = """
{
    "hyperparameters":  {
        "goal": "MAXIMIZE",
        "maxTrials": 6,
        "maxParallelTrials": 3,
        "hyperparameterMetricTag": "accuracy",
        "enableTrialEarlyStopping": True,
        "params": [
            {
                "parameterName": "max_iter",
                "type": "DISCRETE",
                "discreteValues": [500, 1000]
            },
            {
                "parameterName": "alpha",
                "type": "DOUBLE",
                "minValue": 0.0001,
                "maxValue": 0.001,
                "scaleType": "UNIT_LINEAR_SCALE"
            }
        ]
    }
}
"""


# Helper functions
def generate_sampling_query(source_table_name, num_lots, lots):
    """Prepares the data sampling query."""

    sampling_query_template = """
         SELECT *
         FROM 
             `{{ source_table }}` AS cover
         WHERE 
         MOD(ABS(FARM_FINGERPRINT(TO_JSON_STRING(cover))), {{ num_lots }}) IN ({{ lots }})
         """
    query = Template(sampling_query_template).render(
        source_table=source_table_name, num_lots=num_lots, lots=str(lots)[1:-1])

    return query


# Create component factories
component_store = kfp.components.ComponentStore(
    local_search_paths=None, url_search_prefixes=[COMPONENT_URL_SEARCH_PREFIX])

bigquery_query_op = component_store.load_component('bigquery/query')
mlengine_train_op = component_store.load_component('ml_engine/train')
mlengine_deploy_op = component_store.load_component('ml_engine/deploy')
retrieve_best_run_op = func_to_container_op(
    retrieve_best_run, base_image=BASE_IMAGE)
evaluate_model_op = func_to_container_op(evaluate_model, base_image=BASE_IMAGE)


@kfp.dsl.pipeline(
    name='Covertype Classifier Training',
    description='The pipeline training and deploying the Covertype classifierpipeline_yaml'
)
def covertype_train(project_id,
                    region,
                    source_table_name,
                    gcs_root,
                    dataset_id,
                    evaluation_metric_name,
                    evaluation_metric_threshold,
                    model_id,
                    version_id,
                    replace_existing_version,
                    hypertune_settings=HYPERTUNE_SETTINGS,
                    dataset_location='US'):
    """Orchestrates training and deployment of an sklearn model."""

    # Create the training split
    query = generate_sampling_query(
        source_table_name=source_table_name, num_lots=10, lots=[1, 2, 3, 4])

    training_file_path = '{}/{}'.format(gcs_root, TRAINING_FILE_PATH)

    create_training_split = bigquery_query_op(
        query=query,
        project_id=project_id,
        dataset_id=dataset_id,
        table_id='',
        output_gcs_path=training_file_path,
        dataset_location=dataset_location)

    # Create the validation split
    query = generate_sampling_query(
        source_table_name=source_table_name, num_lots=10, lots=[8])

    validation_file_path = '{}/{}'.format(gcs_root, VALIDATION_FILE_PATH)

    create_validation_split = bigquery_query_op(
        query=query,
        project_id=project_id,
        dataset_id=dataset_id,
        table_id='',
        output_gcs_path=validation_file_path,
        dataset_location=dataset_location)

    # Create the testing split
    query = generate_sampling_query(
        source_table_name=source_table_name, num_lots=10, lots=[9])

    testing_file_path = '{}/{}'.format(gcs_root, TESTING_FILE_PATH)

    create_testing_split = bigquery_query_op(
        query=query,
        project_id=project_id,
        dataset_id=dataset_id,
        table_id='',
        output_gcs_path=testing_file_path,
        dataset_location=dataset_location)

    # Tune hyperparameters
    tune_args = [
        '--training_dataset_path',
        create_training_split.outputs['output_gcs_path'],
        '--validation_dataset_path',
        create_validation_split.outputs['output_gcs_path'], '--hptune', 'True'
    ]

    job_dir = '{}/{}/{}'.format(gcs_root, 'jobdir/hypertune',
                                kfp.dsl.RUN_ID_PLACEHOLDER)

    hypertune = mlengine_train_op(
        project_id=project_id,
        region=region,
        master_image_uri=TRAINER_IMAGE,
        job_dir=job_dir,
        args=tune_args,
        training_input=hypertune_settings)

    # Retrieve the best trial
    get_best_trial = retrieve_best_run_op(
            project_id, hypertune.outputs['job_id'])

    # Train the model on a combined training and validation datasets
    job_dir = '{}/{}/{}'.format(gcs_root, 'jobdir', kfp.dsl.RUN_ID_PLACEHOLDER)

    train_args = [
        '--training_dataset_path',
        create_training_split.outputs['output_gcs_path'],
        '--validation_dataset_path',
        create_validation_split.outputs['output_gcs_path'], '--alpha',
        get_best_trial.outputs['alpha'], '--max_iter',
        get_best_trial.outputs['max_iter'], '--hptune', 'False'
    ]

    train_model = mlengine_train_op(
        project_id=project_id,
        region=region,
        master_image_uri=TRAINER_IMAGE,
        job_dir=job_dir,
        args=train_args)

    # Evaluate the model on the testing split
    eval_model = evaluate_model_op(
        dataset_path=str(create_testing_split.outputs['output_gcs_path']),
        model_path=str(train_model.outputs['job_dir']),
        metric_name=evaluation_metric_name)

    # Deploy the model if the primary metric is better than threshold
    with kfp.dsl.Condition(eval_model.outputs['metric_value'] > evaluation_metric_threshold):
        deploy_model = mlengine_deploy_op(
        model_uri=train_model.outputs['job_dir'],
        project_id=project_id,
        model_id=model_id,
        version_id=version_id,
        runtime_version=RUNTIME_VERSION,
        python_version=PYTHON_VERSION,
        replace_existing_version=replace_existing_version)

    # Configure the pipeline to run using the service account defined
    # in the user-gcp-sa k8s secret
    if USE_KFP_SA == 'True':
        kfp.dsl.get_pipeline_conf().add_op_transformer(
              use_gcp_secret('user-gcp-sa'))<jupyter_output>Overwriting ./pipeline/covertype_training_pipeline.py<jupyter_text>The custom components execute in a container image defined in `base_image/Dockerfile`.<jupyter_code>!cat base_image/Dockerfile<jupyter_output>FROM gcr.io/deeplearning-platform-release/base-cpu
RUN pip install -U fire scikit-learn==0.20.4 pandas==0.24.2 kfp==0.2.5<jupyter_text>The training step in the pipeline employes the AI Platform Training component to schedule a  AI Platform Training job in a custom training container. The custom training image is defined in `trainer_image/Dockerfile`.<jupyter_code>!cat trainer_image/Dockerfile<jupyter_output>FROM gcr.io/deeplearning-platform-release/base-cpu
RUN pip install -U fire cloudml-hypertune scikit-learn==0.20.4 pandas==0.24.2
WORKDIR /app
COPY train.py .

ENTRYPOINT ["python", "train.py"]<jupyter_text>Building and deploying the pipelineBefore deploying to AI Platform Pipelines, the pipeline DSL has to be compiled into a pipeline runtime format, also refered to as a pipeline package.  The runtime format is based on [Argo Workflow](https://github.com/argoproj/argo), which is expressed in YAML.   Configure environment settingsUpdate  the below constants  with the settings reflecting your lab environment. - `REGION` - the compute region for AI Platform Training and Prediction- `ARTIFACT_STORE` - the GCS bucket created during installation of AI Platform Pipelines. The bucket name starts with the `hostedkfp-default-` prefix.- `ENDPOINT` - set the `ENDPOINT` constant to the endpoint to your AI Platform Pipelines instance. Then endpoint to the AI Platform Pipelines instance can be found on the [AI Platform Pipelines](https://console.cloud.google.com/ai-platform/pipelines/clusters) page in the Google Cloud Console.1. Open the *SETTINGS* for your instance2. Use the value of the `host` variable in the *Connect to this Kubeflow Pipelines instance from a Python client via Kubeflow Pipelines SKD* section of the *SETTINGS* window.<jupyter_code>REGION = 'us-central1'
ENDPOINT = 'https://71a54605ca951f8a-dot-us-central2.pipelines.googleusercontent.com/'
ARTIFACT_STORE_URI = 'gs://notebooks-project-kubeflowpipelines-default'
PROJECT_ID = !(gcloud config get-value core/project)
PROJECT_ID = PROJECT_ID[0]<jupyter_output><empty_output><jupyter_text>Build the trainer image<jupyter_code>IMAGE_NAME='trainer_image'
TAG='latest'
TRAINER_IMAGE='gcr.io/{}/{}:{}'.format(PROJECT_ID, IMAGE_NAME, TAG)
!gcloud builds submit --timeout 15m --tag $TRAINER_IMAGE trainer_image<jupyter_output>Creating temporary tarball archive of 2 file(s) totalling 3.4 KiB before compression.
Uploading tarball of [trainer_image] to [gs://notebooks-project_cloudbuild/source/1591846773.8-b09a0d7fa65a4fdfbc4faf93d5ebc5d9.tgz]
Created [https://cloudbuild.googleapis.com/v1/projects/notebooks-project/builds/1cf04f3f-5647-4406-89f1-c5c620a0ffcd].
Logs are available at [https://console.cloud.google.com/cloud-build/builds/1cf04f3f-5647-4406-89f1-c5c620a0ffcd?project=57195341408].
----------------------------- REMOTE BUILD OUTPUT ------------------------------
starting build "1cf04f3f-5647-4406-89f1-c5c620a0ffcd"

FETCHSOURCE
Fetching storage object: gs://notebooks-project_cloudbuild/source/1591846773.8-b09a0d7fa65a4fdfbc4faf93d5ebc5d9.tgz#1591846774195452
Copying gs://notebooks-project_cloudbuild/source/1591846773.8-b09a0d7fa65a4fdfbc4faf93d5ebc5d9.tgz#1591846774195452...
/ [1 files][  1.6 KiB/  1.6 KiB]                                                
Operation completed over 1 objects/1.6 KiB.    [...]<jupyter_text>Build the base image for custom components<jupyter_code>IMAGE_NAME='base_image'
TAG='latest'
BASE_IMAGE='gcr.io/{}/{}:{}'.format(PROJECT_ID, IMAGE_NAME, TAG)
!gcloud builds submit --timeout 15m --tag $BASE_IMAGE base_image<jupyter_output>Creating temporary tarball archive of 1 file(s) totalling 122 bytes before compression.
Uploading tarball of [base_image] to [gs://notebooks-project_cloudbuild/source/1591847003.79-5e83bfbd4f9646419f6624ebcd1b6e51.tgz]
Created [https://cloudbuild.googleapis.com/v1/projects/notebooks-project/builds/655fee5f-6db6-444a-8e94-68899cc7baf9].
Logs are available at [https://console.cloud.google.com/cloud-build/builds/655fee5f-6db6-444a-8e94-68899cc7baf9?project=57195341408].
----------------------------- REMOTE BUILD OUTPUT ------------------------------
starting build "655fee5f-6db6-444a-8e94-68899cc7baf9"

FETCHSOURCE
Fetching storage object: gs://notebooks-project_cloudbuild/source/1591847003.79-5e83bfbd4f9646419f6624ebcd1b6e51.tgz#1591847004226672
Copying gs://notebooks-project_cloudbuild/source/1591847003.79-5e83bfbd4f9646419f6624ebcd1b6e51.tgz#1591847004226672...
/ [1 files][  228.0 B/  228.0 B]                                                
Operation completed over 1 objects/228.0 B.  [...]<jupyter_text>Compile the pipelineYou can compile the DSL using an API from the **KFP SDK** or using the **KFP** compiler.To compile the pipeline DSL using the **KFP** compiler.  Set the pipeline's compile time settingsThe pipeline can run using a security context of the GKE default node pool's service account or the service account defined in the `user-gcp-sa` secret of the Kubernetes namespace hosting Kubeflow Pipelines. If you want to use the `user-gcp-sa` service account you change the value of `USE_KFP_SA` to `True`.Note that the default AI Platform Pipelines configuration does not define the `user-gcp-sa` secret.<jupyter_code>USE_KFP_SA = False

COMPONENT_URL_SEARCH_PREFIX = 'https://raw.githubusercontent.com/kubeflow/pipelines/0.2.5/components/gcp/'
RUNTIME_VERSION = '1.15'
PYTHON_VERSION = '3.7'

%env USE_KFP_SA={USE_KFP_SA}
%env BASE_IMAGE={BASE_IMAGE}
%env TRAINER_IMAGE={TRAINER_IMAGE}
%env COMPONENT_URL_SEARCH_PREFIX={COMPONENT_URL_SEARCH_PREFIX}
%env RUNTIME_VERSION={RUNTIME_VERSION}
%env PYTHON_VERSION={PYTHON_VERSION}<jupyter_output>env: USE_KFP_SA=False
env: BASE_IMAGE=gcr.io/notebooks-project/base_image:latest
env: TRAINER_IMAGE=gcr.io/notebooks-project/trainer_image:latest
env: COMPONENT_URL_SEARCH_PREFIX=https://raw.githubusercontent.com/kubeflow/pipelines/0.2.5/components/gcp/
env: RUNTIME_VERSION=1.15
env: PYTHON_VERSION=3.7<jupyter_text>Use the CLI compiler to compile the pipeline<jupyter_code>!dsl-compile --py pipeline/covertype_training_pipeline.py --output covertype_training_pipeline.yaml<jupyter_output><empty_output><jupyter_text>The result is the `covertype_training_pipeline.yaml` file.<jupyter_code>!head covertype_training_pipeline.yaml<jupyter_output>"apiVersion": |-
  argoproj.io/v1alpha1
"kind": |-
  Workflow
"metadata":
  "annotations":
    "pipelines.kubeflow.org/pipeline_spec": |-
      {"description": "The pipeline training and deploying the Covertype classifierpipeline_yaml", "inputs": [{"name": "project_id"}, {"name": "region"}, {"name": "source_table_name"}, {"name": "gcs_root"}, {"name": "dataset_id"}, {"name": "evaluation_metric_name"}, {"name": "evaluation_metric_threshold"}, {"name": "model_id"}, {"name": "version_id"}, {"name": "replace_existing_version"}, {"default": "\n{\n    \"hyperparameters\":  {\n        \"goal\": \"MAXIMIZE\",\n        \"maxTrials\": 6,\n        \"maxParallelTrials\": 3,\n        \"hyperparameterMetricTag\": \"accuracy\",\n        \"enableTrialEarlyStopping\": True,\n        \"params\": [\n            {\n                \"parameterName\": \"max_iter\",\n                \"type\": \"DISCRETE\",\n                \"discreteValues\": [500, 1000]\n            },\n            {\n                \"para[...]<jupyter_text>Deploy the pipeline package<jupyter_code>PIPELINE_NAME='covertype_continuous_training'

!kfp --endpoint $ENDPOINT pipeline upload \
-p $PIPELINE_NAME \
covertype_training_pipeline.yaml<jupyter_output>Pipeline ad9617fc-d3fa-48da-8efa-d73a64bb033b has been submitted

Pipeline Details
------------------
ID           ad9617fc-d3fa-48da-8efa-d73a64bb033b
Name         covertype_continuous_training
Description
Uploaded at  2020-06-11T03:48:32+00:00
+-----------------------------+--------------------------------------------------+
| Parameter Name              | Default Value                                    |
+=============================+==================================================+
| project_id                  |                                                  |
+-----------------------------+--------------------------------------------------+
| region                      |                                                  |
+-----------------------------+--------------------------------------------------+
| source_table_name           |                                                  |
+-----------------------------+--------------------------------------------------+
| gcs_r[...]<jupyter_text>Submitting pipeline runsYou can trigger pipeline runs using an API from the KFP SDK or using KFP CLI. To submit the run using KFP CLI, execute the following commands. Notice how the pipeline's parameters are passed to the pipeline run. List the pipelines in AI Platform Pipelines<jupyter_code>!kfp --endpoint $ENDPOINT pipeline list<jupyter_output>+--------------------------------------+-------------------------------------------------+---------------------------+
| Pipeline ID                          | Name                                            | Uploaded at               |
+======================================+=================================================+===========================+
| ad9617fc-d3fa-48da-8efa-d73a64bb033b | covertype_continuous_training                   | 2020-06-11T03:48:32+00:00 |
+--------------------------------------+-------------------------------------------------+---------------------------+
| 50523945-306c-4383-9437-1e317b9ea1c2 | my_pipeline                                     | 2020-06-04T14:33:58+00:00 |
+--------------------------------------+-------------------------------------------------+---------------------------+
| 8b2535b6-b492-43c2-8dff-069ab1a49161 | [Demo] TFX - Iris classification pipeline       | 2020-06-04T14:03:00+00:00 |
+--------------------------------------+--------[...]<jupyter_text>Submit a runFind the ID of the `covertype_continuous_training` pipeline you uploaded in the previous step and update the value of `PIPELINE_ID` .<jupyter_code>PIPELINE_ID='ad9617fc-d3fa-48da-8efa-d73a64bb033b'
EXPERIMENT_NAME = 'Covertype_Classifier_Training'
RUN_ID = 'Run_001'
SOURCE_TABLE = 'covertype_dataset.covertype'
DATASET_ID = 'splits'
EVALUATION_METRIC = 'accuracy'
EVALUATION_METRIC_THRESHOLD = '0.69'
MODEL_ID = 'covertype_classifier'
VERSION_ID = 'v01'
REPLACE_EXISTING_VERSION = 'True'

GCS_STAGING_PATH = '{}/staging'.format(ARTIFACT_STORE_URI)
!kfp --endpoint $ENDPOINT run submit \
-e $EXPERIMENT_NAME \
-r $RUN_ID \
-p $PIPELINE_ID \
project_id=$PROJECT_ID \
gcs_root=$GCS_STAGING_PATH \
region=$REGION \
source_table_name=$SOURCE_TABLE \
dataset_id=$DATASET_ID \
evaluation_metric_name=$EVALUATION_METRIC \
evaluation_metric_threshold=$EVALUATION_METRIC_THRESHOLD \
model_id=$MODEL_ID \
version_id=$VERSION_ID \
replace_existing_version=$REPLACE_EXISTING_VERSION<jupyter_output>Creating experiment Covertype_Classifier_Training.
Run 67e6ede2-40ad-436b-b5fa-372dfdda406d is submitted
+--------------------------------------+---------+----------+---------------------------+
| run id                               | name    | status   | created at                |
+======================================+=========+==========+===========================+
| 67e6ede2-40ad-436b-b5fa-372dfdda406d | Run_001 |          | 2020-06-11T03:50:07+00:00 |
+--------------------------------------+---------+----------+---------------------------+
{'fraction_non_alphanumeric': 0.1515426497277677, 'fraction_numerical': 0.03457350272232305, 'mean_word_length': 3.1453827346247882, 'pattern_counts': {'":': 43, '<': 42, '<?xml version=': 0, '>': 43, 'https://': 13, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '13', 'dir': 'jupyter-structured-clean-dedup', 'id': '52141', 'license': 'Apache-2.0', 'n_tokens_mistral': 7152, 'n_tokens_neox': 6618, 'n_words': 1483, 'path': 'workshops/kfp-caip-sklearn/lab-02-kfp-pipeline/lab-02.ipynb', 'repo_name': 'bucketdeveloper/mlops-on-gcp', 'size': '22040'}