#@title 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
#
# https://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.

"""
Explanation: Copyright 2020 The TensorFlow Authors.
End of explanation
"""


import sys
if 'google.colab' in sys.modules:
  !pip install --upgrade pip

"""
Explanation: Create a TFX pipeline for your data with Penguin template

Note: We recommend running this tutorial on Google Cloud Vertex AI Workbench. Go to Vertex AI Workbench.
<div class="devsite-table-wrapper"><table class="tfo-notebook-buttons" align="left">
<td><a target="_blank" href="https://www.tensorflow.org/tfx/tutorials/tfx/penguin_template">
<img src="https://www.tensorflow.org/images/tf_logo_32px.png"/>View on TensorFlow.org</a></td>
<td><a target="_blank" href="https://colab.research.google.com/github/tensorflow/tfx/blob/master/docs/tutorials/tfx/penguin_template.ipynb">
<img src="https://www.tensorflow.org/images/colab_logo_32px.png">Run in Google Colab</a></td>
<td><a target="_blank" href="https://github.com/tensorflow/tfx/tree/master/docs/tutorials/tfx/penguin_template.ipynb">
<img width=32px src="https://www.tensorflow.org/images/GitHub-Mark-32px.png">View source on GitHub</a></td>
<td><a href="https://storage.googleapis.com/tensorflow_docs/tfx/docs/tutorials/tfx/penguin_template.ipynb"><img src="https://www.tensorflow.org/images/download_logo_32px.png" />Download notebook</a></td>
</table></div>

Introduction
This document will provide instructions to create a TensorFlow Extended (TFX)
pipeline for your own dataset using penguin template which is provided with
TFX Python package. Created pipeline will be using
Palmer Penguins
dataset initially,
but we will transform the pipeline for your dataset.
Prerequisites

Linux / MacOS
Python 3.6-3.8
Jupyter notebook

Step 1. Copy the predefined template to your project directory.
In this step, we will create a working pipeline project directory and files
by copying files from penguin template in TFX. You can think of this as a
scaffold for your TFX pipeline project.
Update Pip
If we're running in Colab then we should make sure that we have the latest version of Pip.  Local systems can of course be updated separately.
Note: Updating is probably also a good idea if you are running in Vertex AI Workbench.
End of explanation
"""


!pip install -U tfx tensorflow-model-analysis

"""
Explanation: Install required package
First, install TFX and TensorFlow Model Analysis (TFMA).
End of explanation
"""


import tensorflow as tf
import tensorflow_model_analysis as tfma
import tfx

print('TF version: {}'.format(tf.__version__))
print('TFMA version: {}'.format(tfma.__version__))
print('TFX version: {}'.format(tfx.__version__))

"""
Explanation: Let's check the versions of TFX.
End of explanation
"""


PIPELINE_NAME="my_pipeline"
import os
# Set this project directory to your new tfx pipeline project.
PROJECT_DIR=os.path.join(os.path.expanduser("~"), "imported", PIPELINE_NAME)

"""
Explanation: We are ready to create a pipeline.
Set PROJECT_DIR to appropriate destination for your environment.
Default value is ~/imported/${PIPELINE_NAME} which is appropriate for
Google Cloud AI Platform Notebook
environment.
You may give your pipeline a different name by changing the PIPELINE_NAME
below. This will also become the name of the project directory where your
files will be put.
End of explanation
"""


# Set `PATH` to include user python binary directory and a directory containing `skaffold`.
PATH=%env PATH
%env PATH={PATH}:/home/jupyter/.local/bin

!tfx template copy \
  --pipeline-name={PIPELINE_NAME} \
  --destination-path={PROJECT_DIR} \
  --model=penguin

"""
Explanation: Copy template files.
TFX includes the penguin template with the TFX python package.
penguin template
contains many instructions to bring your dataset into the pipeline which is
the purpose of this tutorial.
The tfx template copy CLI command copies predefined template files into your
project directory.
End of explanation
"""


%cd {PROJECT_DIR}

"""
Explanation: Change the working directory context in this notebook to the project directory.
End of explanation
"""


import sys
!{sys.executable} -m models.features_test

"""
Explanation: NOTE: If you are using JupyterLab or Google Cloud AI Platform Notebook,
don't forget to change directory in File Browser on the left by clicking
into the project directory once it is created.

Browse your copied source files
The TFX template provides basic scaffold files to build a pipeline,
including Python source code and sample data. The penguin template uses the
same Palmer Penguins dataset and ML model as the
Penguin example.
Here is brief introduction to each of the Python files.
-   pipeline - This directory contains the definition of the pipeline
    -   configs.py — defines common constants for pipeline runners
    -   pipeline.py — defines TFX components and a pipeline
-   models - This directory contains ML model definitions
    -   features.py, features_test.py — defines features for the model
    -   preprocessing.py, preprocessing_test.py — defines preprocessing
         routines for data
    -   constants.py — defines constants of the model
    -   model.py, model_test.py — defines ML model using ML frameworks
         like TensorFlow
-   local_runner.py — define a runner for local environment which uses
local orchestration engine
-   kubeflow_runner.py — define a runner for Kubeflow Pipelines
orchestration engine
By default, the template only includes standard TFX components. If you need
some customized actions, you can create custom components for your pipeline.
Please see
TFX custom component guide
for the detail.
Unit-test files.
You might notice that there are some files with _test.py in their name.
These are unit tests of the pipeline and it is recommended to add more unit
tests as you implement your own pipelines.
You can run unit tests by supplying the module name of test files with -m
flag. You can usually get a module name by deleting .py extension and
replacing / with ..  For example:
End of explanation
"""


!tfx pipeline create --engine=local --pipeline_path=local_runner.py

"""
Explanation: Create a TFX pipeline in local environment.
TFX supports several orchestration engines to run pipelines. We will use
local orchestration engine. Local orchestration engine runs without any further dependencies, and it is suitable for development and debugging because it runs
on local environment rather than depends on remote computing clusters.
We will use local_runner.py to run your pipeline using local
orchestrator. You have to create a pipeline before running it. You can create
a pipeline with pipeline create command.
End of explanation
"""


# Update and run the pipeline.
!tfx pipeline update --engine=local --pipeline_path=local_runner.py \
 && tfx run create --engine=local --pipeline_name={PIPELINE_NAME}

"""
Explanation: pipeline create command registers your pipeline defined in local_runner.py
without actually running it.
You will run the created pipeline with run create command in following steps.
Step 2. Ingest YOUR data to the pipeline.
The initial pipeline ingests the penguin dataset which is included in the
template. You need to put your data into the pipeline, and most TFX
pipelines start with ExampleGen component.
Choose an ExampleGen
Your data can be stored anywhere your pipeline can access, on either a local or distributed filesystem, or a query-able system. TFX provides various
ExampleGen components
to bring your data into a TFX pipeline. You can choose one from following
example generating components.

CsvExampleGen: Reads CSV files in a directory. Used in
penguin example
and
Chicago taxi example.
ImportExampleGen: Takes TFRecord files with TF Example data format. Used in
MNIST examples.
FileBasedExampleGen for
Avro
or
Parquet
format.
BigQueryExampleGen:
Reads data in Google Cloud BigQuery directly. Used in
Chicago taxi examples.

You can also create your own ExampleGen, for example, tfx includes
a custom ExecampleGen which uses Presto
as a data source. See
the guide
for more information on how to use and develop custom executors.
Once you decide which ExampleGen to use, you will need to modify the pipeline
definition to use your data.

Modify the DATA_PATH in local_runner.py and set it to the location of
your files.
If you have files in local environment, specify the path. This is the
  best option for developing or debugging a pipeline.
If the files are stored in GCS, you can use a path starting with
  gs://{bucket_name}/.... Please make sure that you can access GCS from
  your terminal, for example, using
  gsutil.
  Please follow
  authorization guide in Google Cloud
  if needed.

If you want to use a Query-based ExampleGen like BigQueryExampleGen, you
  need a Query statement to select data from the data source. There are a few
  more things you need to set to use Google Cloud BigQuery as a data source.

In pipeline/configs.py:
Change GOOGLE_CLOUD_PROJECT and GCS_BUCKET_NAME to your GCP
  project and bucket name. The bucket should exist before we run
  the pipeline.
Uncomment BIG_QUERY_WITH_DIRECT_RUNNER_BEAM_PIPELINE_ARGS variable.
Uncomment and set BIG_QUERY_QUERY variable to
  your query statement.
In local_runner.py:
Comment out data_path argument and uncomment query argument
  instead in pipeline.create_pipeline().
In pipeline/pipeline.py:
Comment out data_path argument and uncomment query argument
  in create_pipeline().
Use
  BigQueryExampleGen
  instead of CsvExampleGen.



Replace existing CsvExampleGen to your ExampleGen class in
pipeline/pipeline.py. Each ExampleGen class has different signature.
Please see ExampleGen component guide for more detail. Don't forget to import required modules with
import statements in pipeline/pipeline.py.


The initial pipeline is consist of four components, ExampleGen,
StatisticsGen, SchemaGen and ExampleValidator. We don't need to change
anything for StatisticsGen, SchemaGen and ExampleValidator. Let's run the
pipeline for the first time.
End of explanation
"""


import tensorflow as tf
import tfx
from ml_metadata import errors
from ml_metadata.proto import metadata_store_pb2
from tfx.types import artifact_utils

# TODO(b/171447278): Move these functions into TFX library.

def get_latest_executions(store, pipeline_name, component_id = None):
  """Fetch all pipeline runs."""
  if component_id is None:  # Find entire pipeline runs.
    run_contexts = [
        c for c in store.get_contexts_by_type('run')
        if c.properties['pipeline_name'].string_value == pipeline_name
    ]
  else:  # Find specific component runs.
    run_contexts = [
        c for c in store.get_contexts_by_type('component_run')
        if c.properties['pipeline_name'].string_value == pipeline_name and
           c.properties['component_id'].string_value == component_id
    ]
  if not run_contexts:
    return []
  # Pick the latest run context.
  latest_context = max(run_contexts,
                       key=lambda c: c.last_update_time_since_epoch)
  return store.get_executions_by_context(latest_context.id)

def get_latest_artifacts(store, pipeline_name, component_id = None):
  """Fetch all artifacts from latest pipeline execution."""
  executions = get_latest_executions(store, pipeline_name, component_id)

  # Fetch all artifacts produced from the given executions.
  execution_ids = [e.id for e in executions]
  events = store.get_events_by_execution_ids(execution_ids)
  artifact_ids = [
      event.artifact_id for event in events
      if event.type == metadata_store_pb2.Event.OUTPUT
  ]
  return store.get_artifacts_by_id(artifact_ids)

def find_latest_artifacts_by_type(store, artifacts, artifact_type):
  """Get the latest artifacts of a specified type."""
  # Get type information from MLMD
  try:
    artifact_type = store.get_artifact_type(artifact_type)
  except errors.NotFoundError:
    return []
  # Filter artifacts with type.
  filtered_artifacts = [aritfact for aritfact in artifacts
                        if aritfact.type_id == artifact_type.id]
  # Convert MLMD artifact data into TFX Artifact instances.
  return [artifact_utils.deserialize_artifact(artifact_type, artifact)
      for artifact in filtered_artifacts]


from tfx.orchestration.experimental.interactive import visualizations

def visualize_artifacts(artifacts):
  """Visualizes artifacts using standard visualization modules."""
  for artifact in artifacts:
    visualization = visualizations.get_registry().get_visualization(
        artifact.type_name)
    if visualization:
      visualization.display(artifact)

from tfx.orchestration.experimental.interactive import standard_visualizations
standard_visualizations.register_standard_visualizations()

import pprint

from tfx.orchestration import metadata
from tfx.types import artifact_utils
from tfx.types import standard_artifacts

def preview_examples(artifacts):
  """Preview a few records from Examples artifacts."""
  pp = pprint.PrettyPrinter()
  for artifact in artifacts:
    print("==== Examples artifact:{}({})".format(artifact.name, artifact.uri))
    for split in artifact_utils.decode_split_names(artifact.split_names):
      print("==== Reading from split:{}".format(split))
      split_uri = artifact_utils.get_split_uri([artifact], split)

      # Get the list of files in this directory (all compressed TFRecord files)
      tfrecord_filenames = [os.path.join(split_uri, name)
                            for name in os.listdir(split_uri)]
      # Create a `TFRecordDataset` to read these files
      dataset = tf.data.TFRecordDataset(tfrecord_filenames,
                                        compression_type="GZIP")
      # Iterate over the first 2 records and decode them.
      for tfrecord in dataset.take(2):
        serialized_example = tfrecord.numpy()
        example = tf.train.Example()
        example.ParseFromString(serialized_example)
        pp.pprint(example)

import local_runner

metadata_connection_config = metadata.sqlite_metadata_connection_config(
              local_runner.METADATA_PATH)

"""
Explanation: You should see "Component ExampleValidator is finished." if the pipeline ran successfully.
Examine output of the pipeline.
TFX pipeline produces two kinds of output, artifacts and a
metadata DB(MLMD) which contains
metadata of artifacts and pipeline executions. The location to the output is
defined in local_runner.py. By default, artifacts are stored under
tfx_pipeline_output directory and metadata is stored as an sqlite database
under tfx_metadata directory.
You can use MLMD APIs to examine these outputs. First, we will define some
utility functions to search output artifacts that were just produced.
End of explanation
"""


with metadata.Metadata(metadata_connection_config) as metadata_handler:
    # Search all aritfacts from the previous pipeline run.
    artifacts = get_latest_artifacts(metadata_handler.store, PIPELINE_NAME)
    # Find artifacts of Examples type.
    examples_artifacts = find_latest_artifacts_by_type(
        metadata_handler.store, artifacts,
        standard_artifacts.Examples.TYPE_NAME)
    # Find artifacts generated from StatisticsGen.
    stats_artifacts = find_latest_artifacts_by_type(
        metadata_handler.store, artifacts,
        standard_artifacts.ExampleStatistics.TYPE_NAME)
    # Find artifacts generated from SchemaGen.
    schema_artifacts = find_latest_artifacts_by_type(
        metadata_handler.store, artifacts,
        standard_artifacts.Schema.TYPE_NAME)
    # Find artifacts generated from ExampleValidator.
    anomalies_artifacts = find_latest_artifacts_by_type(
        metadata_handler.store, artifacts,
        standard_artifacts.ExampleAnomalies.TYPE_NAME)

"""
Explanation: Now we can read metadata of output artifacts from MLMD.
End of explanation
"""


preview_examples(examples_artifacts)

"""
Explanation: Now we can examine outputs from each component.
Tensorflow Data Validation(TFDV)
is used in StatisticsGen, SchemaGen and ExampleValidator, and TFDV can
be used to visualize outputs from these components.
In this tutorial, we will use visualzation helper methods in TFX which use TFDV
internally to show the visualization. Please see
TFX components tutorial
to learn more about each component.
Examine output form ExampleGen
Let's examine output from ExampleGen. Take a look at the first two examples for
each split:
End of explanation
"""


visualize_artifacts(stats_artifacts)

"""
Explanation: By default, TFX ExampleGen divides examples into two splits, train and
eval, but you can
adjust your split configuration.
Examine output from StatisticsGen
End of explanation
"""


visualize_artifacts(schema_artifacts)

"""
Explanation: These statistics are supplied to SchemaGen to construct a schema of data
automatically.
Examine output from SchemaGen
End of explanation
"""


visualize_artifacts(anomalies_artifacts)

"""
Explanation: This schema is automatically inferred from the output of StatisticsGen.
We will use this generated schema in this tutorial, but you also can
modify and customize the schema.
Examine output from ExampleValidator
End of explanation
"""


!tfx pipeline update --engine=local --pipeline_path=local_runner.py \
 && tfx run create --engine=local --pipeline_name={PIPELINE_NAME}

"""
Explanation: If any anomalies were found, you may review your data that all examples
follow your assumptions. Outputs from other components like StatistcsGen might
be useful. Found anomalies don't block the pipeline execution.
You can see the available features from the outputs of the SchemaGen. If
your features can be used to construct ML model in Trainer directly, you can
skip the next step and go to Step 4. Otherwise you can do some feature
engineering work in the next step. Transform component is needed when
full-pass operations like calculating averages are required, especially when
you need to scale.
Step 3. (Optional) Feature engineering with Transform component.
In this step, you will define various feature engineering job which will be
used by Transform component in the pipeline. See
Transform component guide
for more information.
This is only necessary if you training code requires additional feature(s)
which is not available in the output of ExampleGen. Otherwise, feel free to
fast forward to next step of using Trainer.
Define features of the model
models/features.py contains constants to define features for the model
including feature names, size of vocabulariy and so on. By default penguin
template has two costants, FEATURE_KEYS and LABEL_KEY, because our penguin
model solves a classification problem using supervised learning and all
features are continuous numeric features. See
feature definitions from the chicago taxi example
for another example.
Implement preprocessing for training / serving in preprocessing_fn().
Actual feature engineering happens in preprocessing_fn() function in
models/preprocessing.py.
In preprocessing_fn you can define a series of functions that manipulate the
input dict of tensors to produce the output dict of tensors. There are helper
functions like scale_to_0_1 and compute_and_apply_vocabulary in the
TensorFlow Transform API or you can simply use regular TensorFlow functions.
By default penguin template includes example usages of
tft.scale_to_z_score
function to normalize feature values.
See Tensflow Transform guide
for more information about authoring preprocessing_fn.
Add Transform component to the pipeline.
If your preprocessing_fn is ready, add Transform component to the pipeline.

In pipeline/pipeline.py file, uncomment # components.append(transform)
to add the component to the pipeline.

You can update the pipeline and run again.
End of explanation
"""


with metadata.Metadata(metadata_connection_config) as metadata_handler:
    # Search all aritfacts from the previous run of Transform component.
    artifacts = get_latest_artifacts(metadata_handler.store,
                                     PIPELINE_NAME, "Transform")
    # Find artifacts of Examples type.
    transformed_examples_artifacts = find_latest_artifacts_by_type(
        metadata_handler.store, artifacts,
        standard_artifacts.Examples.TYPE_NAME)

preview_examples(transformed_examples_artifacts)

"""
Explanation: If the pipeline ran successfully, you should see "Component Transform is
finished." somewhere in the log. Because Transform component and
ExampleValidator component are not dependent to each other, the order of
executions is not fixed. That said, either of Transform and
ExampleValidator can be the last component in the pipeline execution.
Examine output from Transform
Transform component creates two kinds of outputs, a Tensorflow graph and
transformed examples. The transformed examples are Examples artifact type which
is also produced by ExampleGen, but this one contains transformed feature
values instead.
You can examine them as we did in the previous step.
End of explanation
"""


!tfx pipeline update --engine=local --pipeline_path=local_runner.py \
 && tfx run create --engine=local --pipeline_name={PIPELINE_NAME}

"""
Explanation: Step 4. Train your model with Trainer component.
We will build a ML model using Trainer component. See
Trainer component guide
for more information. You need to provide your model code to the Trainer
component.
Define your model.
In penguin template, models.model.run_fn is used as run_fn argument for
Trainer component. It means that run_fn() function in models/model.py
will be called when Trainer component runs. You can see the code to construct
a simple DNN model using keras API in given code. See
TensorFlow 2.x in TFX
guide for more information about using keras API in TFX.
In this run_fn, you should build a model and save it to a directory pointed
by fn_args.serving_model_dir which is specified by the component. You can use
other arguments in fn_args which is passed into the run_fn. See
related codes
for the full list of arguments in fn_args.
Define your features in models/features.py and use them as needed. If you
have transformed your features in Step 3, you should use transformed features
as inputs to your model.
Add Trainer component to the pipeline.
If your run_fn is ready, add Trainer component to the pipeline.

In pipeline/pipeline.py file, uncomment # components.append(trainer)
to add the component to the pipeline.

Arguments for the trainer component might depends on whether you use Transform
component or not.
- If you do NOT use Transform component, you don't need to change the
arguments.
- If you use Transform component, you need to change arguments
when creating a Trainer component instance.
  - Change examples argument to
  examples=transform.outputs['transformed_examples'],. We need to use
  transformed examples for training.
  - Add transform_graph argument like
  transform_graph=transform.outputs['transform_graph'],. This graph
  contains TensorFlow graph for the transform operations.
  - After above changes, the code for Trainer component creation will
  look like following.
python
  # If you use a Transform component.
  trainer = Trainer(
      run_fn=run_fn,
      examples=transform.outputs['transformed_examples'],
      transform_graph=transform.outputs['transform_graph'],
      schema=schema_gen.outputs['schema'],
      ...
You can update the pipeline and run again.
End of explanation
"""


# Update and run the pipeline.
!tfx pipeline update --engine=local --pipeline_path=local_runner.py \
 && tfx run create --engine=local --pipeline_name={PIPELINE_NAME}

"""
Explanation: When this execution runs successfully, you have now created and run your first
TFX pipeline for your model. Congratulations!
Your new model will be located in some place under the output directory, but it
would be better to have a model in fixed location or service outside of the TFX
pipeline which holds many interim results. Even better with continuous
evaluation of the built model which is critical in ML production systems. We
will see how continuous evaluation and deployments work in TFX in the next step.
Step 5.  (Optional) Evaluate the model with Evaluator and publish with pusher.
Evaluator component
continuously evaluate every built model from Trainer, and
Pusher copies the model to
a predefined location in the file system or even to
Google Cloud AI Platform Models.
Adds Evaluator component to the pipeline.
In pipeline/pipeline.py file:
1. Uncomment # components.append(model_resolver) to add latest model resolver
to the pipeline. Evaluator can be used to compare a model with old baseline
model which passed Evaluator in last pipeline run. LatestBlessedModelResolver
finds the latest model which passed Evaluator.
1. Set proper tfma.MetricsSpec for your model. Evaluation might be different
for every ML model. In the penguin template, SparseCategoricalAccuracy was used
because we are solving a multi category classification problem. You also need
to specify tfma.SliceSpec to analyze your model for specific slices. For more
detail, see
Evaluator component guide.
1. Uncomment # components.append(evaluator) to add the component to the
pipeline.
You can update the pipeline and run again.
End of explanation
"""


# Install TFMA notebook extension.
!jupyter labextension install tensorflow_model_analysis@{tfma.__version__}

"""
Explanation: Examine output of Evaluator
This step requires TensorFlow Model Analysis(TFMA) Jupyter notebook extension.
Note that the version of the TFMA notebook extension should be identical to the
version of TFMA python package.
Following command will install TFMA notebook extension from NPM registry. It
might take several minutes to complete.
End of explanation
"""


with metadata.Metadata(metadata_connection_config) as metadata_handler:
  # Search all aritfacts from the previous pipeline run.
  artifacts = get_latest_artifacts(metadata_handler.store, PIPELINE_NAME)
  model_evaluation_artifacts = find_latest_artifacts_by_type(
      metadata_handler.store, artifacts,
      standard_artifacts.ModelEvaluation.TYPE_NAME)

if model_evaluation_artifacts:
  tfma_result = tfma.load_eval_result(model_evaluation_artifacts[0].uri)
  tfma.view.render_slicing_metrics(tfma_result)

"""
Explanation: If installation is completed, please reload your browser to make the
extension take effect.
End of explanation
"""


# Update and run the pipeline.
!tfx pipeline update --engine=local --pipeline_path=local_runner.py \
 && tfx run create --engine=local --pipeline_name={PIPELINE_NAME}

"""
Explanation: Adds Pusher component to the pipeline.
If the model looks promising, we need to publish the model.
Pusher component
can publish the model to a location in the filesystem or to GCP AI Platform
Models using
a custom executor.
Evaluator component continuously evaluate every built model from Trainer,
and Pusher copies the model to
a predefined location in the file system or even to
Google Cloud AI Platform Models.

In local_runner.py, set SERVING_MODEL_DIR to a directory to publish.
In pipeline/pipeline.py file, uncomment # components.append(pusher)
to add Pusher to the pipeline.

You can update the pipeline and run again.
End of explanation
"""


!pip install --upgrade -q kfp

# Download skaffold and set it executable.
!curl -Lo skaffold https://storage.googleapis.com/skaffold/releases/latest/skaffold-linux-amd64 && chmod +x skaffold

"""
Explanation: You should be able to find your new model at SERVING_MODEL_DIR.
Step 6.  (Optional) Deploy your pipeline to Kubeflow Pipelines on GCP.
As mentioned earlier, local_runner.py is good for debugging or development
purpose but not a best solution for production workloads. In this step, we will
deploy the pipeline to Kubeflow Pipelines on Google Cloud.
Preparation
We need kfp python package and skaffold program to deploy a pipeline to a
Kubeflow Pipelines cluster.
End of explanation
"""


# Move skaffold binary into your path
!mv skaffold /home/jupyter/.local/bin/

"""
Explanation: You need to move skaffold binary to the place where your shell can find it.
Or you can specify the path to skaffold when you run tfx binary with
--skaffold-cmd flag.
End of explanation
"""


ENDPOINT='' # Enter your ENDPOINT here.

"""
Explanation: You also need a Kubeflow Pipelines cluster to run the pipeline. Please
follow Step 1 and 2 in
TFX on Cloud AI Platform Pipelines tutorial.
When your cluster is ready, open the pipeline dashboard by clicking
Open Pipelines Dashboard in the
Pipelines page of the Google cloud console.
The URL of this page is ENDPOINT to request a pipeline run. The endpoint
value is everything in the URL after the https://, up to, and including,
googleusercontent.com. Put your endpoint to following code block.
End of explanation
"""


# Read GCP project id from env.
shell_output=!gcloud config list --format 'value(core.project)' 2>/dev/null
GOOGLE_CLOUD_PROJECT=shell_output[0]

# Docker image name for the pipeline image.
CUSTOM_TFX_IMAGE='gcr.io/' + GOOGLE_CLOUD_PROJECT + '/tfx-pipeline'

"""
Explanation: To run our code in a Kubeflow Pipelines cluster, we need to pack our code into
a container image. The image will be built automatically while deploying our
pipeline, and you only need to set a name and an container registry for your
image. In our example, we will use
Google Container registry,
and name it tfx-pipeline.
End of explanation
"""


!gsutil cp data/data.csv gs://{GOOGLE_CLOUD_PROJECT}-kubeflowpipelines-default/tfx-template/data/penguin/

"""
Explanation: Set data location.
Your data should be accessible from the Kubeflow Pipelines cluster. If you have
used data in your local environment, you might need to upload it to remote
storage like Google Cloud Storage. For example, we can upload penguin data to a
default bucket which is created automatically when a Kubeflow Pipelines cluster
is deployed like following.
End of explanation
"""


!tfx pipeline create  \
--engine=kubeflow \
--pipeline-path=kubeflow_runner.py \
--endpoint={ENDPOINT} \
--build-target-image={CUSTOM_TFX_IMAGE}

"""
Explanation: Update the data location stored at DATA_PATH in kubeflow_runner.py.
If you are using BigQueryExampleGen, there is no need to upload the data file,
but please make sure that kubeflow_runner.py uses the same query and
beam_pipeline_args argument for pipeline.create_pipeline() function.
Deploy the pipeline.
If everything is ready, you can create a pipeline using tfx pipeline create
command.

Note: When creating a pipeline for Kubeflow Pipelines, we need a container
image which will be used to run our pipeline. And skaffold will build the
image for us. Because skaffold pulls base images from the docker hub, it will
take 5~10 minutes when we build the image for the first time, but it will take
much less time from the second build.
End of explanation
"""


!tfx run create --engine=kubeflow --pipeline-name={PIPELINE_NAME} --endpoint={ENDPOINT}

"""
Explanation: Now start an execution run with the newly created pipeline using the
tfx run create command.
End of explanation
"""


