ETL from Cloud Storage or S3.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Copyright 2019 Google Inc. All Rights Reserved.\n",
    "#\n",
    "# Licensed under the Apache License, Version 2.0 (the \"License\");\n",
    "# you may not use this file except in compliance with the License.\n",
    "# You may obtain a copy of the License at\n",
    "#\n",
    "#     http://www.apache.org/licenses/LICENSE-2.0\n",
    "#\n",
    "# Unless required by applicable law or agreed to in writing, software\n",
    "# distributed under the License is distributed on an \"AS IS\" BASIS,\n",
    "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
    "# See the License for the specific language governing permissions and\n",
    "# limitations under the License."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "tags": [
     "parameters"
    ]
   },
   "outputs": [],
   "source": [
    "# Install Pipeline SDK - This only needs to be ran once in the enviroment. \n",
    "# you can find the latest package @ https://github.com/kubeflow/pipelines/releases\n",
    "#KFP_PACKAGE = 'https://storage.googleapis.com/ml-pipeline/release/0.1.20/kfp.tar.gz'\n",
    "#!pip3 install $KFP_PACKAGE --upgrade"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "EXPERIMENT_NAME = 'Hellow world!'\n",
    "GCLOUD_SDK = 'google/cloud-sdk:latest'\n",
    "GCS_PATH = 'gs://chavoshi-dev-mlpipeline'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ETL from Google Cloud Storage or S3 \n",
    "\n",
    "This section demonstrates how you can access content from cloud based blog storage such as GCS and S3 to a pod during execution of the pipeline. Reference documentation is as follows. \n",
    "* https://cloud.google.com/storage/docs/quickstart-gsutil\n",
    "* https://cloud.google.com/storage/docs/reference/libraries\n",
    "\n",
    "\n",
    "Note that apart from using Google Specific libraries native Tensorflow io libraries can read and write to GCS & S3. \n",
    "* https://github.com/tensorflow/examples/blob/master/community/en/docs/deploy/s3.md\n",
    "\n",
    "\n",
    "Finally many common libraries such as Pandas have built support for access common cloud storage such as S3 and GCS and Bigquery. \n",
    "* https://cloud.google.com/bigquery/docs/pandas-gbq-migration"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "import kfp\n",
    "import kfp.dsl as dsl\n",
    "from kfp.gcp import use_gcp_secret\n",
    "from kubernetes import client as k8s_client\n",
    "from kfp import compiler\n",
    "from kfp import notebook\n",
    "from kfp import components as comp\n",
    "from kfp.onprem import mount_pvc"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Accessing Google Cloud storage\n",
    "There are four methods that can be used to access Google cloud storage. \n",
    "\n",
    "* Using Gsutil\n",
    "* Using Google Cloud Storage API  #TODO\n",
    "* Using Tensorflow IO operations  #TODO\n",
    "* Using application specific slution such as Pandas Library integration #TODO\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "def gcs_using_gsutil_op(gcs_path:str):\n",
    "    return dsl.ContainerOp(\n",
    "        name = 'ETL using Gsutil',\n",
    "        image = GCLOUD_SDK,\n",
    "        command=['sh', '-c'],\n",
    "        arguments = [  '''gsutil ls ''' +  gcs_path ]\n",
    "    ).apply(use_gcp_secret('user-gcp-sa'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Define the pipeline\n",
    "Pipeline function has to be decorated with the `@dsl.pipeline` decorator"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "import kfp.dsl as dsl\n",
    "@dsl.pipeline(\n",
    "   name='ETL',\n",
    "   description='A toy pipeline demonstrates accessing blob storage'\n",
    ")\n",
    "def etl_pipeline():\n",
    "    #Creating a one step pipeline\n",
    "    gcs_using_gsutil = gcs_using_gsutil_op(GCS_PATH) #Returns a dsl.ContainerOp class instance. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Compile the pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline_func = etl_pipeline\n",
    "pipeline_filename = pipeline_func.__name__ + '.pipeline.zip'\n",
    "import kfp.compiler as compiler\n",
    "compiler.Compiler().compile(pipeline_func, pipeline_filename)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Submit the pipeline for execution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "Experiment link <a href=\"/pipeline/#/experiments/details/571db2d0-a74a-4170-a68c-42f2498819bb\" target=\"_blank\" >here</a>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "#Get or create an experiment and submit a pipeline run\n",
    "import kfp\n",
    "client = kfp.Client()\n",
    "experiment = client.create_experiment(EXPERIMENT_NAME)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "Run link <a href=\"/pipeline/#/runs/details/129b288c-a412-11e9-a9d6-42010a800018\" target=\"_blank\" >here</a>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "#Specify pipeline argument values\n",
    "arguments = {}\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
}