{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Develop Kubeflow Pipeline in Notebook\n",
    "\n",
    "This notebook shows how to build, compile and run a Kubeflow pipeline using Kubeflow SDK in Jupyter notebooks.\n",
    "\n",
    "This notebook will make sure you understand \n",
    "- Develop a Kubeflow pipeline with the KFP SDK\n",
    "- Submit Kubeflow pipelines using the KFP SDK\n",
    "- Monitor Kubeflow pipeline running status\n",
    "\n",
    "For KFP SDK details, please check documentaion [SDK Overview](https://www.kubeflow.org/docs/pipelines/sdk/sdk-overview/)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Import kfp sdk"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import kfp\n",
    "import kfp.dsl as dsl\n",
    "from kfp import compiler"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Develop your Kubeflow Pipeline "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create pipeline component from a python function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def add_op(a, b):\n",
    "    \n",
    "    return dsl.ContainerOp(\n",
    "        name='Add Operation',\n",
    "        image='python:3.6.5',\n",
    "        command=['sh', '-c'],\n",
    "        arguments=['python -c \"print($0+$1)\" | tee $2', str(a), str(b), '/tmp/output'],\n",
    "        file_outputs={\n",
    "            'data': '/tmp/output',\n",
    "        }\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Build a pipeline using the component"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@dsl.pipeline(\n",
    "   name='Calculation pipeline',\n",
    "   description='A toy pipeline that performs arithmetic calculations.'\n",
    ")\n",
    "def calc_pipeline(\n",
    "   a=0,\n",
    "   b=7,\n",
    "   c=17,\n",
    "):\n",
    "    #Passing pipeline parameter and a constant value as operation arguments\n",
    "    add_task = add_op(a, 4) #Returns a dsl.ContainerOp class instance. \n",
    "    \n",
    "    #You can create explicit dependency between the tasks using xyz_task.after(abc_task)\n",
    "    add_2_task = add_op(a, b)\n",
    "    \n",
    "    add_3_task = add_op(add_task.output, add_2_task.output)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Compile and run the pipeline\n",
    "\n",
    "Create an Experiment in the Pipeline System"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline_func = calc_pipeline\n",
    "pipeline_filename = pipeline_func.__name__ + '.pipeline.zip'\n",
    "compiler.Compiler().compile(pipeline_func, pipeline_filename)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create an Experiment in the Pipeline System\n",
    "\n",
    "Pipeline system requires an \"Experiment\" to group pipeline runs. You can create a new experiment, or call client.list_experiments() to get existing ones.\n",
    "\n",
    "You can also use `default` Experiment to host your pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get or create an experiment and submit a pipeline run\n",
    "EXPERIMENT_NAME='add_operation'\n",
    "client = kfp.Client()\n",
    "experiment = client.create_experiment(EXPERIMENT_NAME)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Submit the pipeline for execution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Specify pipeline argument values\n",
    "arguments = {'a': 7, 'b': 8}\n",
    "\n",
    "# Submit a pipeline run\n",
    "run_name = pipeline_func.__name__ + ' run'\n",
    "run_result = client.run_pipeline(experiment.id, run_name, pipeline_filename, arguments)\n",
    "\n",
    "# This link leads to the run information page. \n",
    "# Note: There is a bug in JupyterLab that modifies the URL and makes the link stop working"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "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.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}