index.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Databolt Flow\n",
    "For data scientists and data engineers, d6tflow is a python library which makes building complex data science workflows easy, fast and intuitive.\n",
    "\n",
    "https://github.com/d6t/d6tflow"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Benefits of using d6tflow\n",
    "\n",
    "[4 Reasons Why Your Machine Learning Code is Probably Bad](https://medium.com/@citynorman/4-reasons-why-your-machine-learning-code-is-probably-bad-c291752e4953)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Example Usage For a Machine Learning Workflow\n",
    "\n",
    "Below is an example of a typical machine learning workflow: you retreive data, preprocess it, train a model and evaluate the model output.\n",
    "\n",
    "In this example you will:\n",
    "* Build a machine learning workflow made up of individual tasks\n",
    "* Check task dependencies and their execution status\n",
    "* Execute the model training task including dependencies\n",
    "* Save intermediary task output to Parquet, pickle and in-memory\n",
    "* Load task output to pandas dataframe and model object for model evaluation\n",
    "* Intelligently rerun workflow after changing a preprocessing parameter\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 1: Define workflow"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Welcome to d6tflow!\n"
     ]
    }
   ],
   "source": [
    "import d6tflow\n",
    "import luigi\n",
    "from luigi.util import inherits\n",
    "import sklearn, sklearn.datasets, sklearn.svm, sklearn.linear_model\n",
    "import pandas as pd\n",
    "\n",
    "# define workflow\n",
    "class TaskGetData(d6tflow.tasks.TaskPqPandas):  # save dataframe as parquet\n",
    "\n",
    "    def run(self):\n",
    "        ds = sklearn.datasets.load_breast_cancer()\n",
    "        df_train = pd.DataFrame(ds.data, columns=ds.feature_names)\n",
    "        df_train['y'] = ds.target\n",
    "        self.save(df_train) # quickly save dataframe\n",
    "\n",
    "\n",
    "@d6tflow.requires(TaskGetData) # define dependency\n",
    "class TaskPreprocess(d6tflow.tasks.TaskPqPandas):\n",
    "    do_preprocess = luigi.BoolParameter(default=True) # parameter for preprocessing yes/no\n",
    "\n",
    "    def run(self):\n",
    "        df_train = self.input().load() # quickly load required data\n",
    "        if self.do_preprocess:\n",
    "            df_train.iloc[:,:-1] = sklearn.preprocessing.scale(df_train.iloc[:,:-1])\n",
    "        self.save(df_train)\n",
    "\n",
    "@d6tflow.requires(TaskPreprocess) # automatically pass parameters upstream\n",
    "class TaskTrain(d6tflow.tasks.TaskPickle): # save output as pickle\n",
    "    model = luigi.Parameter(default='ols') # parameter for model selection\n",
    "\n",
    "    def run(self):\n",
    "        df_train = self.input().load()\n",
    "        if self.model=='ols':\n",
    "            model = sklearn.linear_model.LogisticRegression()\n",
    "        elif self.model=='svm':\n",
    "            model = sklearn.svm.SVC()\n",
    "        else:\n",
    "            raise ValueError('invalid model selection')\n",
    "        model.fit(df_train.drop('y',1), df_train['y'])\n",
    "        self.save(model)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 2: Define experiments"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# goal: compare performance of two models\n",
    "params_model1 = {'do_preprocess':True, 'model':'ols'}\n",
    "params_model2 = {'do_preprocess':False, 'model':'svm'}\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 3: Run workflow"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "===== Luigi Execution Summary =====\n",
      "\n",
      "Scheduled 3 tasks of which:\n",
      "* 3 ran successfully:\n",
      "    - 1 TaskGetData()\n",
      "    - 1 TaskPreprocess(do_preprocess=True)\n",
      "    - 1 TaskTrain(do_preprocess=True, model=ols)\n",
      "\n",
      "This progress looks :) because there were no failed tasks or missing dependencies\n",
      "\n",
      "===== Luigi Execution Summary =====\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "LuigiRunResult(status=<LuigiStatusCode.SUCCESS: (':)', 'there were no failed tasks or missing dependencies')>,worker=<luigi.worker.Worker object at 0x0000015DF1351D08>,scheduling_succeeded=True)"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# run workflow for model 1\n",
    "d6tflow.run(TaskTrain(**params_model1)) \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      " ===== Luigi Execution Preview ===== \n",
      "\n",
      "\n",
      "└─--[TaskTrain-{'do_preprocess': 'False', 'model': 'svm'} (\u001b[94mPENDING\u001b[0m)]\n",
      "   └─--[TaskPreprocess- (\u001b[94mPENDING\u001b[0m)]\n",
      "      └─--[TaskGetData- (\u001b[92mCOMPLETE\u001b[0m)]\n",
      "\n",
      " ===== Luigi Execution Preview ===== \n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Intelligently rerun workflow after changing parameters\n",
    "d6tflow.preview(TaskTrain(**params_model2))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "===== Luigi Execution Summary =====\n",
      "\n",
      "Scheduled 3 tasks of which:\n",
      "* 1 complete ones were encountered:\n",
      "    - 1 TaskGetData()\n",
      "* 2 ran successfully:\n",
      "    - 1 TaskPreprocess(do_preprocess=False)\n",
      "    - 1 TaskTrain(do_preprocess=False, model=svm)\n",
      "\n",
      "This progress looks :) because there were no failed tasks or missing dependencies\n",
      "\n",
      "===== Luigi Execution Summary =====\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "LuigiRunResult(status=<LuigiStatusCode.SUCCESS: (':)', 'there were no failed tasks or missing dependencies')>,worker=<luigi.worker.Worker object at 0x0000015DF67C74C8>,scheduling_succeeded=True)"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# run workflow for model 2\n",
    "d6tflow.run(TaskTrain(**params_model2))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0.9876977152899824\n",
      "0.9226713532513181\n"
     ]
    }
   ],
   "source": [
    "# compare results from new model\n",
    "# Load task output to pandas dataframe and model object for model evaluation\n",
    "\n",
    "model1 = TaskTrain(**params_model1).output().load()\n",
    "df_train = TaskPreprocess(**params_model1).output().load()\n",
    "print(model1.score(df_train.drop('y',1), df_train['y']))\n",
    "# 0.987\n",
    "\n",
    "model2 = TaskTrain(**params_model2).output().load()\n",
    "df_train = TaskPreprocess(**params_model2).output().load()\n",
    "print(model2.score(df_train.drop('y',1), df_train['y']))\n",
    "# 0.922\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Next steps: Transition code to d6tflow\n",
    "\n",
    "See https://d6tflow.readthedocs.io/en/latest/transition.html"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}