training_time_series_forecasting.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "kfp 1.1.2\n",
      "CPython 3.6.9\n",
      "IPython 7.11.1\n"
     ]
    }
   ],
   "source": [
    "import kfp\n",
    "import kfp.components as comp\n",
    "from kfp.components import InputPath, OutputPath\n",
    "import kfp.dsl as dsl\n",
    "from kfp.aws import use_aws_secret\n",
    "from typing import NamedTuple\n",
    "from itertools import product\n",
    "\n",
    "%load_ext watermark\n",
    "%watermark -v -iv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'experiments': [{'created_at': datetime.datetime(2020, 12, 16, 7, 31, 48, tzinfo=tzlocal()),\n",
       "                  'description': None,\n",
       "                  'id': '2e84ecc3-6228-4675-906f-a6e538a7ab1c',\n",
       "                  'name': 'Default',\n",
       "                  'resource_references': [{'key': {'id': 'anonymous',\n",
       "                                                   'type': 'NAMESPACE'},\n",
       "                                           'name': None,\n",
       "                                           'relationship': 'OWNER'}],\n",
       "                  'storage_state': 'STORAGESTATE_AVAILABLE'}],\n",
       " 'next_page_token': None,\n",
       " 'total_size': 1}"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# In v1.1.0, in-cluster communication from notebook to Kubeflow Pipeline is not supported in this phase.\n",
    "# In order to use kfp as previous, user needs to pass a cookie to KFP for communication as a workaround.\n",
    "# https://www.kubeflow.org/docs/aws/pipeline/#authenticate-kubeflow-pipeline-using-sdk-inside-cluster\n",
    "\n",
    "authservice_session='authservice_session=<cookie>'\n",
    "client = kfp.Client(host='http://192.168.52.92:30178/pipeline', cookies=authservice_session)\n",
    "namespace='anonymous'\n",
    "client.list_experiments(namespace=namespace)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Component: Load Raw Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "def load_raw_data(path: str, raw_data_path: OutputPath('CSV')):\n",
    "    print(raw_data_path)\n",
    "\n",
    "    import subprocess\n",
    "\n",
    "    # downlaod the dataset from the mlflow repo\n",
    "    def download_dataset():\n",
    "        subprocess.call(['apt-get', 'update'])\n",
    "        subprocess.call(['apt-get', 'install', 'curl', '-y'])\n",
    "        subprocess.call(['curl', '-o', 'household_power_consumption.txt',\n",
    "                         'https://raw.githubusercontent.com/felix-exel/mlflow/master/household_power_consumption.txt'])\n",
    "\n",
    "    download_dataset()\n",
    "\n",
    "    import pandas as pd\n",
    "\n",
    "    df = pd.read_csv(path, sep=';',\n",
    "                     parse_dates={'dt': ['Date', 'Time']}, infer_datetime_format=True,\n",
    "                     low_memory=False, na_values=['nan', '?'], index_col='dt')\n",
    "\n",
    "    df.to_csv(raw_data_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "load_raw_data_op = comp.func_to_container_op(load_raw_data,\n",
    "                                             base_image='python:3.7-slim',\n",
    "                                             packages_to_install=['pandas==1.0.5'])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Component: Filling NaNs with Mean"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "def filling_nans_with_mean(input_data_path: InputPath('CSV'), output_data_path: OutputPath('CSV')):\n",
    "\n",
    "    import pandas as pd\n",
    "\n",
    "    df = pd.read_csv(input_data_path, sep=',', header=0, infer_datetime_format=True,\n",
    "                     low_memory=False, parse_dates=True, index_col='dt')\n",
    "\n",
    "    # filling nan with mean in any columns\n",
    "    for j in range(0, df.shape[1]):\n",
    "        df.iloc[:, j] = df.iloc[:, j].fillna(df.iloc[:, j].mean())\n",
    "\n",
    "    print(df.isnull().sum())\n",
    "\n",
    "    df.to_csv(output_data_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "filling_nans_with_mean_op = comp.func_to_container_op(filling_nans_with_mean,\n",
    "                                                      base_image='python:3.7-slim',\n",
    "                                                      packages_to_install=['pandas==1.0.5'])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Component: Split Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "def split_data(input_data_path: InputPath('CSV'), output_train_data_path: OutputPath('CSV'), output_test_data_path: OutputPath('CSV')):\n",
    "\n",
    "    import pandas as pd\n",
    "\n",
    "    df = pd.read_csv(input_data_path, sep=',', header=0, infer_datetime_format=True,\n",
    "                     low_memory=False, parse_dates=True, index_col='dt')\n",
    "\n",
    "    # split training and test set\n",
    "    number_training = round(df.shape[0] * 0.8)\n",
    "\n",
    "    train_data = df[:number_training]\n",
    "    test_data = df[number_training:]\n",
    "\n",
    "    train_data.to_csv(output_train_data_path)\n",
    "    test_data.to_csv(output_test_data_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "split_data_op = comp.func_to_container_op(split_data,\n",
    "                                          base_image='python:3.7-slim',\n",
    "                                          packages_to_install=['pandas==1.0.5'])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Component: Standardization"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "def standardization(input_train_data_path: InputPath('CSV'), input_test_data_path: InputPath('CSV'),\n",
    "                    output_train_data_path: OutputPath('CSV'), output_test_data_path: OutputPath('CSV')):\n",
    "\n",
    "    import pandas as pd\n",
    "\n",
    "    train_data = pd.read_csv(input_train_data_path, sep=',', header=0, infer_datetime_format=True,\n",
    "                             low_memory=False, parse_dates=True, index_col='dt')\n",
    "\n",
    "    test_data = pd.read_csv(input_test_data_path, sep=',', header=0, infer_datetime_format=True,\n",
    "                            low_memory=False, parse_dates=True, index_col='dt')\n",
    "\n",
    "    # Standardization\n",
    "    mean = train_data.mean(axis=0)\n",
    "    std = train_data.std(axis=0)\n",
    "    train_data_standardized = (train_data - mean) / std\n",
    "\n",
    "    # take mean and std of the train data to standardize the test data\n",
    "    test_data_standardized = (test_data - mean) / std\n",
    "\n",
    "    print(train_data_standardized.describe())\n",
    "\n",
    "    train_data_standardized.to_csv(output_train_data_path)\n",
    "    test_data_standardized.to_csv(output_test_data_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "standardization_op = comp.func_to_container_op(standardization,\n",
    "                                               base_image='python:3.7-slim',\n",
    "                                               packages_to_install=['pandas==1.0.5'])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Component: Model Training"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "def training(input_train_data_path: InputPath('CSV'),\n",
    "             input_test_data_path: InputPath('CSV'),\n",
    "             batch_size: int,\n",
    "             window_length: int,\n",
    "             future_length: int,\n",
    "             dropout_fc: float,\n",
    "             hidden_layer_size: int,\n",
    "             n_output_features: int,\n",
    "             mlpipeline_metrics_path: OutputPath('Metrics')):\n",
    "\n",
    "    import pandas as pd\n",
    "    import tensorflow as tf\n",
    "    import tensorflow.keras as keras\n",
    "    import numpy as np\n",
    "    import json\n",
    "    import mlflow\n",
    "    import os\n",
    "\n",
    "    def set_mlflow_settings():\n",
    "        os.environ[\"GIT_PYTHON_REFRESH\"] = \"quiet\"\n",
    "\n",
    "        registry_uri = 'mysql+pymysql://mlflow:mlflow@mysql.mlflow.svc.cluster.local:3306/mlflow'\n",
    "        tracking_uri = 'http://mlflow-service.mlflow.svc.cluster.local:5001'\n",
    "\n",
    "        mlflow.tracking.set_registry_uri(registry_uri)\n",
    "        mlflow.tracking.set_tracking_uri(tracking_uri)\n",
    "\n",
    "    set_mlflow_settings()\n",
    "\n",
    "    train_data = pd.read_csv(input_train_data_path, sep=',', header=0, infer_datetime_format=True,\n",
    "                             low_memory=False, parse_dates=True, index_col='dt')\n",
    "\n",
    "    test_data = pd.read_csv(input_test_data_path, sep=',', header=0, infer_datetime_format=True,\n",
    "                            low_memory=False, parse_dates=True, index_col='dt')\n",
    "\n",
    "    def build_lstm_2_layer_model(window_length=50, future_length=1, n_input_features=7,\n",
    "                                 n_output_features=3, units_lstm_layer=30, dropout_rate=0.2):\n",
    "        \"\"\"Builds 2 Layer LSTM-based TF Model in functional API.\n",
    "        Args:\n",
    "          window_length: Input Data as Numpy Array, Shape (rows, n_features)\n",
    "          future_length: Number of time steps that will be predicted in the future.\n",
    "          n_input_features: Number of features that will be used as Input.\n",
    "          n_output_features: Number of features that will be predicted.\n",
    "          units_lstm_layer: Number of Neurons for the LSTM Layers.\n",
    "          dropout_rate: Dropout Rate for the last Fully Connected Dense Layer.\n",
    "        Returns:\n",
    "          keras.models.Model\n",
    "        \"\"\"\n",
    "        inputs = keras.layers.Input(\n",
    "            shape=[window_length, n_input_features], dtype=np.float32)\n",
    "\n",
    "        # Layer1\n",
    "        lstm1_output, lstm1_state_h, lstm1_state_c = keras.layers.LSTM(units=units_lstm_layer, return_state=True,\n",
    "                                                                       return_sequences=True)(inputs)\n",
    "        lstm1_state = [lstm1_state_h, lstm1_state_c]\n",
    "\n",
    "        # Layer 2\n",
    "        lstm2_output, lstm2_state_h, lstm2_state_c = keras.layers.LSTM(units=units_lstm_layer, return_state=True,\n",
    "                                                                       return_sequences=True)(lstm1_output,\n",
    "                                                                                              initial_state=lstm1_state)\n",
    "\n",
    "        reshaped = tf.reshape(lstm2_output,\n",
    "                              [-1, window_length * units_lstm_layer])\n",
    "        # Dropout\n",
    "        dropout = tf.keras.layers.Dropout(dropout_rate)(reshaped)\n",
    "\n",
    "        fc_layer = keras.layers.Dense(n_output_features * future_length, kernel_initializer='he_normal', dtype=tf.float32)(\n",
    "            dropout)\n",
    "\n",
    "        output = tf.reshape(fc_layer,\n",
    "                            [-1, future_length, n_output_features])\n",
    "\n",
    "        model = keras.models.Model(inputs=[inputs],\n",
    "                                   outputs=[output])\n",
    "        return model\n",
    "\n",
    "    def apply_sliding_window_tf_data_api(train_data_x, batch_size=64, window_length=50, future_length=1,\n",
    "                                         n_output_features=1, validate=False, shift=1):\n",
    "        \"\"\"Applies sliding window on the fly by using the TF Data API.\n",
    "        Args:\n",
    "          train_data_x: Input Data as Numpy Array, Shape (rows, n_features)\n",
    "          batch_size: Batch Size.\n",
    "          window_length: Window Length or Window Size.\n",
    "          future_length: Number of time steps that will be predicted in the future.\n",
    "          n_output_features: Number of features that will be predicted.\n",
    "          validate: True if input data is a validation set and does not need to be shuffled\n",
    "          shift: Shifts the Sliding Window by this Parameter.\n",
    "        Returns:\n",
    "          tf.data.Dataset\n",
    "        \"\"\"\n",
    "        def make_window_dataset(ds, window_size=5, shift=1, stride=1):\n",
    "            windows = ds.window(window_size, shift=shift, stride=stride)\n",
    "\n",
    "            def sub_to_batch(sub):\n",
    "                return sub.batch(window_size, drop_remainder=True)\n",
    "\n",
    "            windows = windows.flat_map(sub_to_batch)\n",
    "            return windows\n",
    "\n",
    "        X = tf.data.Dataset.from_tensor_slices(train_data_x.astype(np.float32))\n",
    "        y = tf.data.Dataset.from_tensor_slices(\n",
    "            train_data_x[window_length:, :n_output_features].astype(np.float32))\n",
    "        ds_x = make_window_dataset(\n",
    "            X, window_size=window_length, shift=shift, stride=1)\n",
    "        ds_y = make_window_dataset(\n",
    "            y, window_size=future_length, shift=shift, stride=1)\n",
    "\n",
    "        if not validate:\n",
    "            train_tf_data = tf.data.Dataset.zip((ds_x, ds_y)).cache() \\\n",
    "                .shuffle(buffer_size=200000, reshuffle_each_iteration=True).batch(batch_size).prefetch(5)\n",
    "            return train_tf_data\n",
    "        else:\n",
    "            return tf.data.Dataset.zip((ds_x, ds_y)).batch(batch_size).prefetch(1)\n",
    "\n",
    "    class MlflowLogging(tf.keras.callbacks.Callback):\n",
    "        def __init__(self, **kwargs):\n",
    "            super().__init__()  # handles base args (e.g., dtype)\n",
    "\n",
    "        def on_epoch_end(self, epoch, logs=None):\n",
    "            keys = list(logs.keys())\n",
    "            for key in keys:\n",
    "                mlflow.log_metric(str(key), logs.get(key), step=epoch)\n",
    "\n",
    "    # enable gpu growth if gpu is available\n",
    "    gpu_devices = tf.config.experimental.list_physical_devices('GPU')\n",
    "    for device in gpu_devices:\n",
    "        tf.config.experimental.set_memory_growth(device, True)\n",
    "\n",
    "    with mlflow.start_run() as run:\n",
    "        # log parameter\n",
    "        mlflow.log_param('batch_size', batch_size)\n",
    "        mlflow.log_param('window_length', window_length)\n",
    "        mlflow.log_param('hidden_layer_size', hidden_layer_size)\n",
    "        mlflow.log_param('dropout_fc_layer', dropout_fc)\n",
    "        mlflow.log_param('future_length', future_length)\n",
    "        mlflow.log_param('n_output_features', n_output_features)\n",
    "\n",
    "        model = build_lstm_2_layer_model(window_length=window_length,\n",
    "                                         future_length=future_length,\n",
    "                                         n_output_features=n_output_features,\n",
    "                                         units_lstm_layer=hidden_layer_size,\n",
    "                                         dropout_rate=dropout_fc)\n",
    "\n",
    "        train_data_sliding_window = apply_sliding_window_tf_data_api(train_data.values,\n",
    "                                                                     batch_size=batch_size,\n",
    "                                                                     window_length=window_length,\n",
    "                                                                     future_length=future_length,\n",
    "                                                                     n_output_features=n_output_features)\n",
    "\n",
    "        test_data_sliding_window = apply_sliding_window_tf_data_api(test_data.values,\n",
    "                                                                    batch_size=batch_size,\n",
    "                                                                    window_length=window_length,\n",
    "                                                                    future_length=future_length,\n",
    "                                                                    n_output_features=n_output_features,\n",
    "                                                                    validate=True)\n",
    "\n",
    "        model.compile(loss='mse', optimizer=keras.optimizers.Nadam(learning_rate=1e-3),\n",
    "                      metrics=['mse', 'mae'])\n",
    "\n",
    "        model.fit(train_data_sliding_window, shuffle=True, initial_epoch=0, epochs=10, validation_data=test_data_sliding_window,\n",
    "                  callbacks=[MlflowLogging()])\n",
    "\n",
    "        metrics = model.evaluate(test_data_sliding_window)\n",
    "        print(metrics)\n",
    "\n",
    "        # Exports MSE and MAE metrics:\n",
    "        metrics = {\n",
    "            'metrics': [{\n",
    "                'name': 'mse',\n",
    "                'numberValue':  float(metrics[1]),\n",
    "            }, {\n",
    "                'name': 'mae',\n",
    "                'numberValue':  float(metrics[2]),\n",
    "            }]}\n",
    "\n",
    "        with open(mlpipeline_metrics_path, 'w') as f:\n",
    "            json.dump(metrics, f)\n",
    "\n",
    "        # Save the model to the artifact store: s3 bucket\n",
    "        model_path = './model'\n",
    "        os.makedirs(model_path)\n",
    "        model.save(model_path)\n",
    "\n",
    "        mlflow.tensorflow.log_model(tf_saved_model_dir=model_path,\n",
    "                                    tf_meta_graph_tags='serve',\n",
    "                                    tf_signature_def_key='serving_default',\n",
    "                                    artifact_path='saved_model',\n",
    "                                    registered_model_name='Electric Power Consumption Forecasting')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "training_op = comp.func_to_container_op(training,\n",
    "                                        base_image='tensorflow/tensorflow:2.3.1',\n",
    "                                        packages_to_install=['pandas>=1.0.5', 'mlflow', 'boto3', 'pymysql'])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating a Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "@dsl.pipeline(\n",
    "    name='Training pipeline',\n",
    "    description='Training pipeline for time series forecasting on household power consumption dataset.'\n",
    "\n",
    ")\n",
    "def training_pipeline(\n",
    "    path='./household_power_consumption.txt'\n",
    "):\n",
    "\n",
    "    # Returns a dsl.ContainerOp class instance.\n",
    "    load_raw_data_task = load_raw_data_op(\n",
    "        path).set_display_name('Load Raw Data')\n",
    "\n",
    "    key_output = str(list(load_raw_data_task.outputs.keys())[0])\n",
    "\n",
    "    filling_nans_with_mean_op_task = filling_nans_with_mean_op(load_raw_data_task.outputs[key_output]).after(\n",
    "        load_raw_data_task).set_display_name('Filling NaNs with Mean')\n",
    "\n",
    "    key_output = str(list(filling_nans_with_mean_op_task.outputs.keys())[0])\n",
    "\n",
    "    split_data_task = split_data_op(filling_nans_with_mean_op_task.outputs[key_output]).after(\n",
    "        filling_nans_with_mean_op_task).set_display_name('Split Data')\n",
    "\n",
    "    key_output1 = str(list(split_data_task.outputs.keys())[0])\n",
    "    key_output2 = str(list(split_data_task.outputs.keys())[1])\n",
    "\n",
    "    standardization_task = standardization_op(split_data_task.outputs[key_output1], split_data_task.outputs[key_output2]).after(\n",
    "        split_data_task).set_display_name('Standardization')\n",
    "\n",
    "    key_output1 = str(list(standardization_task.outputs.keys())[0])\n",
    "    key_output2 = str(list(standardization_task.outputs.keys())[1])\n",
    "\n",
    "    grid_search_dic = {'hidden_layer_size': [20, 40],\n",
    "                       'batch_size': [64],\n",
    "                       'future_length': [5],\n",
    "                       'window_length': [50],\n",
    "                       'dropout_fc': [0.0, 0.2],\n",
    "                       'n_output_features': [1]}\n",
    "\n",
    "    # Cartesian product\n",
    "    grid_search_param = [dict(zip(grid_search_dic, v))\n",
    "                         for v in product(*grid_search_dic.values())]\n",
    "\n",
    "    for i, params in enumerate(grid_search_param):\n",
    "        batch_size = params['batch_size']\n",
    "        window_length = params['window_length']\n",
    "        future_length = params['future_length']\n",
    "        dropout_fc = params['dropout_fc']\n",
    "        hidden_layer_size = params['hidden_layer_size']\n",
    "        n_output_features = params['n_output_features']\n",
    "\n",
    "        training_task = training_op(standardization_task.outputs[key_output1],\n",
    "                                    standardization_task.outputs[key_output2],\n",
    "                                    batch_size,\n",
    "                                    window_length,\n",
    "                                    future_length,\n",
    "                                    dropout_fc,\n",
    "                                    hidden_layer_size,\n",
    "                                    n_output_features).after(standardization_task).set_display_name('Model Training '+str(i+1))\\\n",
    "            .apply(use_aws_secret('aws-secret', 'AWS_ACCESS_KEY_ID', 'AWS_SECRET_ACCESS_KEY'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating a Pipeline Run"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<a href=\"http://192.168.52.92:30178/pipeline/#/experiments/details/2e84ecc3-6228-4675-906f-a6e538a7ab1c\" target=\"_blank\" >Experiment details</a>."
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "<a href=\"http://192.168.52.92:30178/pipeline/#/runs/details/0613dbf9-d870-4477-b4b3-549edff88e16\" target=\"_blank\" >Run details</a>."
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "RunPipelineResult(run_id=0613dbf9-d870-4477-b4b3-549edff88e16)"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "arguments = {'path': './household_power_consumption.txt'}\n",
    "\n",
    "# Submit a pipeline run\n",
    "client.create_run_from_pipeline_func(\n",
    "    training_pipeline, arguments=arguments, namespace=namespace)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Uploading the Pipeline to be reuseable by others"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<a href=http://192.168.52.92:30178/pipeline/#/pipelines/details/1d6a0b49-b620-40c8-9c2a-e9b0d9860ac0>Pipeline details</a>."
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "{'created_at': datetime.datetime(2020, 12, 16, 8, 3, 40, tzinfo=tzlocal()),\n",
       " 'default_version': {'code_source_url': None,\n",
       "                     'created_at': datetime.datetime(2020, 12, 16, 8, 3, 40, tzinfo=tzlocal()),\n",
       "                     'id': '1d6a0b49-b620-40c8-9c2a-e9b0d9860ac0',\n",
       "                     'name': 'Training Pipeline .',\n",
       "                     'package_url': None,\n",
       "                     'parameters': [{'name': 'path',\n",
       "                                     'value': './household_power_consumption.txt'}],\n",
       "                     'resource_references': [{'key': {'id': '1d6a0b49-b620-40c8-9c2a-e9b0d9860ac0',\n",
       "                                                      'type': 'PIPELINE'},\n",
       "                                              'name': None,\n",
       "                                              'relationship': 'OWNER'}]},\n",
       " 'description': None,\n",
       " 'error': None,\n",
       " 'id': '1d6a0b49-b620-40c8-9c2a-e9b0d9860ac0',\n",
       " 'name': 'Training Pipeline .',\n",
       " 'parameters': [{'name': 'path', 'value': './household_power_consumption.txt'}],\n",
       " 'url': None}"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "kfp.compiler.Compiler().compile(training_pipeline, 'workflow.yaml')\n",
    "client.upload_pipeline(pipeline_package_path='workflow.yaml',\n",
    "                             pipeline_name='Training Pipeline .')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "kubeflow_notebook": {
   "docker_image": "gcr.io/arrikto-public/tensorflow-1.15.2-notebook-cpu:1.0.0.arr1",
   "experiment": {
    "id": "",
    "name": ""
   },
   "experiment_name": "",
   "pipeline_description": "",
   "pipeline_name": "",
   "volumes": [
    {
     "annotations": [],
     "mount_point": "/home/jovyan",
     "name": "workspace-server-k451mcy01",
     "size": 5,
     "size_type": "Gi",
     "snapshot": false,
     "type": "clone"
    }
   ]
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}