Intro_to_Luigi.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Flow-based Programming with Luigi for doing Science\n",
    "\n",
    "A better way to make + manage data pipelines"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Harry Jack\n",
    "harry.jack@outlook.com\n",
    "\n",
    "https://github.com/hazbottles/luigi_talk\n",
    "\n",
    "\n",
    "- 2013: B. Sc. (Mathematics) & B. Eng. (Chemical Engineering), __The University of Sydney__\n",
    "- 2014 - 2017: Meteorologist / Forecast System Scientist, __Australian Bureau of Meteorology__\n",
    "- 2018 - Present: Modeller / Lead Modeller, __Solcast__\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Data Pipelines\n",
    "\n",
    "## Imperative Programming\n",
    "\n",
    "```\n",
    "data = []\n",
    "dates = ['2019-04-01', '2019-04-02']\n",
    "for date in dates:\n",
    "    input1 = get_input1_data(date)\n",
    "    output1 = make_output0(date, input1)\n",
    "    \n",
    "    input2 = get_input2_data(date)\n",
    "    output2 = make_output1(date, input2, output1)\n",
    "    \n",
    "    data.append([output1, output2])\n",
    "    \n",
    "# debugging + recovering from errors?\n",
    "# starting from half-done?\n",
    "# making new outputs?\n",
    "```\n",
    "\n",
    "## Flow-based Programming\n",
    "\n",
    "- specify Tasks with inputs+outputs\n",
    "- let the framework decide how to execute the code\n",
    "\n",
    "`luigi --module examples.py MakeData --first 2019-04-09 --last 2019-04-10`\n",
    "\n",
    "![dependency graph](./example_dependency_graph.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "## How can this make my data-pipelines better?\n",
    "\n",
    "\n",
    "- modularise-ability\n",
    "- reproduce-ability\n",
    "- extend-ability\n",
    "- port-ability\n",
    "- paralellise-ability"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "\n",
    "## Luigi\n",
    "\n",
    "https://github.com/spotify/luigi\n",
    "\n",
    "https://luigi.readthedocs.io/en/latest/\n",
    "\n",
    "### Installation\n",
    "\n",
    "`pip install luigi`\n",
    "\n",
    "`conda install luigi`\n",
    "\n",
    "### Alternatives\n",
    "\n",
    "- dask\n",
    "- celery\n",
    "- airflow"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Python 3.7.3\r\n"
     ]
    }
   ],
   "source": [
    "! python -V"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Name: luigi\r\n",
      "Version: 2.8.3\r\n",
      "Summary: Workflow mgmgt + task scheduling + dependency resolution\r\n",
      "Home-page: https://github.com/spotify/luigi\r\n",
      "Author: The Luigi Authors\r\n",
      "Author-email: None\r\n",
      "License: Apache License 2.0\r\n",
      "Location: /home/harry/miniconda3/envs/luigi_talk/lib/python3.7/site-packages\r\n",
      "Requires: tornado, python-dateutil, python-daemon\r\n",
      "Required-by: \r\n"
     ]
    }
   ],
   "source": [
    "! pip show luigi"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style>.container { width:100% !important; }</style>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "from IPython.core.display import display, HTML\n",
    "display(HTML(\"<style>.container { width:100% !important; }</style>\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "import datetime\n",
    "import time\n",
    "import os\n",
    "\n",
    "import luigi"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "# make somewhere to dump luigi output\n",
    "! rm -rf /tmp/luigi_talk_examples/\n",
    "! mkdir /tmp/luigi_talk_examples/"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Basic Luigi Task"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "task: DayName(date=2019-04-09)\n",
      "task.complete(): False\n",
      "\n",
      "target: <luigi.local_target.LocalTarget object at 0x7f5a602a1668>\n",
      "target.path: /tmp/luigi_talk_examples/DayName_2019-04-09.txt\n",
      "target.exists(): False\n"
     ]
    }
   ],
   "source": [
    "class DayName(luigi.Task):\n",
    "    date = luigi.DateParameter() # cast to a datetime.date\n",
    "    \n",
    "    def output(self):\n",
    "        return luigi.LocalTarget(\n",
    "            # ISO-8601 date format on pain of death\n",
    "            f'/tmp/luigi_talk_examples/DayName_{self.date:%Y-%m-%d}.txt'\n",
    "        )\n",
    "            \n",
    "    def run(self):\n",
    "            \n",
    "        # simulate taking some time to process\n",
    "        time.sleep(5)\n",
    "        result = self.date.strftime('%A')  # self.date is a datetime.date\n",
    "        \n",
    "        # Number 1 Luigi gotcha: don't stream directly to output file!\n",
    "        tmpfile = f'{self.output().path}.tmp'\n",
    "        with open(tmpfile, 'w') as fd:\n",
    "            fd.write(result)\n",
    "        # make the output atomically\n",
    "        os.rename(tmpfile, self.output().path)\n",
    "            \n",
    "# inspect the Task object\n",
    "task = DayName(date=datetime.date(2019,4,9))\n",
    "target = task.output()\n",
    "print(f'task: {repr(task)}')\n",
    "print(f'task.complete(): {task.complete()}')\n",
    "print('')\n",
    "print(f'target: {repr(target)}')\n",
    "print(f'target.path: {target.path}')\n",
    "print(f'target.exists(): {target.exists()}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO: Informed scheduler that task   DayName_2019_04_09_2e0ba28ae6   has status   PENDING\n",
      "INFO: Done scheduling tasks\n",
      "INFO: Running Worker with 1 processes\n",
      "INFO: [pid 74] Worker Worker(salt=376876960, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) running   DayName(date=2019-04-09)\n",
      "INFO: [pid 74] Worker Worker(salt=376876960, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) done      DayName(date=2019-04-09)\n",
      "INFO: Informed scheduler that task   DayName_2019_04_09_2e0ba28ae6   has status   DONE\n",
      "INFO: Worker Worker(salt=376876960, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) was stopped. Shutting down Keep-Alive thread\n",
      "INFO: \n",
      "===== Luigi Execution Summary =====\n",
      "\n",
      "Scheduled 1 tasks of which:\n",
      "* 1 ran successfully:\n",
      "    - 1 DayName(date=2019-04-09)\n",
      "\n",
      "This progress looks :) because there were no failed tasks or missing dependencies\n",
      "\n",
      "===== Luigi Execution Summary =====\n",
      "\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "True\n"
     ]
    }
   ],
   "source": [
    "# 1. check if tasks are complete `task.complete()`\n",
    "# 2. schedule tasks which are not complete\n",
    "# 3. run tasks `task.run()`\n",
    "\n",
    "# command line: `luigi --module examples DayName --date 2019-04-09 --local-scheduler`\n",
    "luigi.build(\n",
    "    [DayName(date=datetime.date(2019,4,9))],\n",
    "    local_scheduler=True,\n",
    ")\n",
    "print(DayName(date=datetime.date(2019,4,9)).complete())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DayName_2019-04-09.txt\r\n"
     ]
    }
   ],
   "source": [
    "! ls /tmp/luigi_talk_examples"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Tuesday"
     ]
    }
   ],
   "source": [
    "! cat /tmp/luigi_talk_examples/DayName_2019-04-09.txt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO: Informed scheduler that task   DayName_2019_04_09_2e0ba28ae6   has status   DONE\n",
      "INFO: Done scheduling tasks\n",
      "INFO: Running Worker with 1 processes\n",
      "INFO: Worker Worker(salt=706858153, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) was stopped. Shutting down Keep-Alive thread\n",
      "INFO: \n",
      "===== Luigi Execution Summary =====\n",
      "\n",
      "Scheduled 1 tasks of which:\n",
      "* 1 complete ones were encountered:\n",
      "    - 1 DayName(date=2019-04-09)\n",
      "\n",
      "Did not run any tasks\n",
      "This progress looks :) because there were no failed tasks or missing dependencies\n",
      "\n",
      "===== Luigi Execution Summary =====\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# 1. check if tasks are complete `task.complete()`\n",
    "# 2. schedule tasks which are not complete\n",
    "# 3. run tasks `task.run()`\n",
    "\n",
    "# Task output already exists, so it isn't run!\n",
    "\n",
    "# luigi --module examples DayName --date 2019-04-09 --local-scheduler\n",
    "luigi.build(\n",
    "    [DayName(date=datetime.date(2019,4,9))],\n",
    "    local_scheduler=True,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Importance of atomicity"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO: Informed scheduler that task   DayNameErrorful_2019_04_09_2e0ba28ae6   has status   PENDING\n",
      "INFO: Done scheduling tasks\n",
      "INFO: Running Worker with 1 processes\n",
      "INFO: [pid 74] Worker Worker(salt=574673838, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) running   DayNameErrorful(date=2019-04-09)\n",
      "ERROR: [pid 74] Worker Worker(salt=574673838, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) failed    DayNameErrorful(date=2019-04-09)\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/harry/miniconda3/envs/luigi_talk/lib/python3.7/site-packages/luigi/worker.py\", line 199, in run\n",
      "    new_deps = self._run_get_new_deps()\n",
      "  File \"/home/harry/miniconda3/envs/luigi_talk/lib/python3.7/site-packages/luigi/worker.py\", line 139, in _run_get_new_deps\n",
      "    task_gen = self.task.run()\n",
      "  File \"<ipython-input-10-2becc3ea137d>\", line 12, in run\n",
      "    fd.write(self.date('%A'))  # will error but makes an empty file\n",
      "TypeError: 'datetime.date' object is not callable\n",
      "INFO: Informed scheduler that task   DayNameErrorful_2019_04_09_2e0ba28ae6   has status   FAILED\n",
      "INFO: Worker Worker(salt=574673838, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) was stopped. Shutting down Keep-Alive thread\n",
      "INFO: \n",
      "===== Luigi Execution Summary =====\n",
      "\n",
      "Scheduled 1 tasks of which:\n",
      "* 1 failed:\n",
      "    - 1 DayNameErrorful(date=2019-04-09)\n",
      "\n",
      "This progress looks :( because there were failed tasks\n",
      "\n",
      "===== Luigi Execution Summary =====\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "False"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "class DayNameErrorful(luigi.Task):\n",
    "    date = luigi.DateParameter() # a datetime.date\n",
    "    \n",
    "    def output(self):\n",
    "        return luigi.LocalTarget(\n",
    "            f'/tmp/luigi_talk_examples/DayNameErrorful_{self.date:%Y-%m-%d}.txt'\n",
    "        )\n",
    "            \n",
    "    def run(self):\n",
    "        time.sleep(3)\n",
    "        with open(self.output().path, 'w') as fd:\n",
    "            fd.write(self.date('%A'))  # will error but makes an empty file\n",
    "\n",
    "\n",
    "# note: you get a proper stack trace\n",
    "luigi.build(\n",
    "    [DayNameErrorful(date=datetime.date(2019,4,9))],\n",
    "    local_scheduler=True,\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "total 0\r\n",
      "-rw-rw-rw- 1 harry harry 7 Apr  9 16:38 DayName_2019-04-09.txt\r\n",
      "-rw-rw-rw- 1 harry harry 0 Apr  9 16:38 DayNameErrorful_2019-04-09.txt\r\n"
     ]
    }
   ],
   "source": [
    "# we've made an empty file by mistake\n",
    "! ls /tmp/luigi_talk_examples/ -lh"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO: Informed scheduler that task   DayNameErrorful_2019_04_09_2e0ba28ae6   has status   DONE\n",
      "INFO: Done scheduling tasks\n",
      "INFO: Running Worker with 1 processes\n",
      "INFO: Worker Worker(salt=942812875, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) was stopped. Shutting down Keep-Alive thread\n",
      "INFO: \n",
      "===== Luigi Execution Summary =====\n",
      "\n",
      "Scheduled 1 tasks of which:\n",
      "* 1 complete ones were encountered:\n",
      "    - 1 DayNameErrorful(date=2019-04-09)\n",
      "\n",
      "Did not run any tasks\n",
      "This progress looks :) because there were no failed tasks or missing dependencies\n",
      "\n",
      "===== Luigi Execution Summary =====\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Future runs think it has succeeded!\n",
    "luigi.build(\n",
    "    [DayNameErrorful(date=datetime.date(2019,4,9))],\n",
    "    local_scheduler=True,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Task Dependencies"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "class DayNameFrench(luigi.Task):\n",
    "    date = luigi.DateParameter()\n",
    "    \n",
    "    def requires(self):\n",
    "        return DayName(date=self.date)\n",
    "    \n",
    "    def output(self):\n",
    "        return luigi.LocalTarget(f'/tmp/luigi_talk_examples/DayNameFrench_{self.date:%Y-%m-%d}.txt')\n",
    "\n",
    "    def run(self):\n",
    "        with open(self.input().path, 'r') as fd:\n",
    "            english = fd.read()\n",
    "        french = f\"Oh la la c'est le {english}\"\n",
    "        time.sleep(5)\n",
    "        \n",
    "        tmpfile = f'{self.output().path}.tmp'\n",
    "        with open(tmpfile, 'w') as fd:\n",
    "            fd.write(french)\n",
    "        os.rename(tmpfile, self.output().path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DayName_2019-04-09.txt\tDayNameErrorful_2019-04-09.txt\r\n"
     ]
    }
   ],
   "source": [
    "! ls /tmp/luigi_talk_examples/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO: Informed scheduler that task   DayNameFrench_2019_04_09_2e0ba28ae6   has status   PENDING\n",
      "INFO: Informed scheduler that task   DayName_2019_04_09_2e0ba28ae6   has status   DONE\n",
      "INFO: Done scheduling tasks\n",
      "INFO: Running Worker with 1 processes\n",
      "INFO: [pid 74] Worker Worker(salt=709494939, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) running   DayNameFrench(date=2019-04-09)\n",
      "INFO: [pid 74] Worker Worker(salt=709494939, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) done      DayNameFrench(date=2019-04-09)\n",
      "INFO: Informed scheduler that task   DayNameFrench_2019_04_09_2e0ba28ae6   has status   DONE\n",
      "INFO: Worker Worker(salt=709494939, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) was stopped. Shutting down Keep-Alive thread\n",
      "INFO: \n",
      "===== Luigi Execution Summary =====\n",
      "\n",
      "Scheduled 2 tasks of which:\n",
      "* 1 complete ones were encountered:\n",
      "    - 1 DayName(date=2019-04-09)\n",
      "* 1 ran successfully:\n",
      "    - 1 DayNameFrench(date=2019-04-09)\n",
      "\n",
      "This progress looks :) because there were no failed tasks or missing dependencies\n",
      "\n",
      "===== Luigi Execution Summary =====\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Uses output of DayName(2019-04-09)\n",
    "# i.e. we pick up the computation half-way through, without having to write special logic to do so!\n",
    "\n",
    "# luigi --module examples DayNameFrench --date 2019-04-09 --local-scheduler\n",
    "luigi.build([DayNameFrench(date=datetime.date(2019,4,9))], local_scheduler=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DayName_2019-04-09.txt\t\tDayNameFrench_2019-04-09.txt\r\n",
      "DayNameErrorful_2019-04-09.txt\r\n"
     ]
    }
   ],
   "source": [
    "! ls /tmp/luigi_talk_examples/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO: Informed scheduler that task   DayNameFrench_2019_04_08_f7ae138f92   has status   PENDING\n",
      "INFO: Informed scheduler that task   DayName_2019_04_08_f7ae138f92   has status   PENDING\n",
      "INFO: Done scheduling tasks\n",
      "INFO: Running Worker with 1 processes\n",
      "INFO: [pid 74] Worker Worker(salt=701873738, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) running   DayName(date=2019-04-08)\n",
      "INFO: [pid 74] Worker Worker(salt=701873738, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) done      DayName(date=2019-04-08)\n",
      "INFO: Informed scheduler that task   DayName_2019_04_08_f7ae138f92   has status   DONE\n",
      "INFO: [pid 74] Worker Worker(salt=701873738, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) running   DayNameFrench(date=2019-04-08)\n",
      "INFO: [pid 74] Worker Worker(salt=701873738, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) done      DayNameFrench(date=2019-04-08)\n",
      "INFO: Informed scheduler that task   DayNameFrench_2019_04_08_f7ae138f92   has status   DONE\n",
      "INFO: Worker Worker(salt=701873738, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) was stopped. Shutting down Keep-Alive thread\n",
      "INFO: \n",
      "===== Luigi Execution Summary =====\n",
      "\n",
      "Scheduled 2 tasks of which:\n",
      "* 2 ran successfully:\n",
      "    - 1 DayName(date=2019-04-08)\n",
      "    - 1 DayNameFrench(date=2019-04-08)\n",
      "\n",
      "This progress looks :) because there were no failed tasks or missing dependencies\n",
      "\n",
      "===== Luigi Execution Summary =====\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Runs both DayName(2019-04-08) and DayNameFrench(2019-04-08)\n",
    "\n",
    "# luigi --module examples DayNameFrench --date 2019-04-08 --local-scheduler\n",
    "luigi.build([DayNameFrench(date=datetime.date(2019,4,8))], local_scheduler=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DayName_2019-04-08.txt\t\tDayNameFrench_2019-04-08.txt\r\n",
      "DayName_2019-04-09.txt\t\tDayNameFrench_2019-04-09.txt\r\n",
      "DayNameErrorful_2019-04-09.txt\r\n"
     ]
    }
   ],
   "source": [
    "! ls /tmp/luigi_talk_examples/"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Multiple Dependencies"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "# Dependencies on multiple tasks\n",
    "\n",
    "def iter_days(first, last):\n",
    "    while first <= last:\n",
    "        yield first\n",
    "        first += datetime.timedelta(days=1)\n",
    "\n",
    "class SummaryCsv(luigi.Task):\n",
    "    first = luigi.DateParameter()\n",
    "    last = luigi.DateParameter()\n",
    "    \n",
    "    def requires(self):\n",
    "        return {\n",
    "            date: [DayName(date), DayNameFrench(date)]\n",
    "            for date in iter_days(self.first, self.last)\n",
    "        }\n",
    "    \n",
    "    def output(self):\n",
    "        return luigi.LocalTarget(\n",
    "            f'/tmp/luigi_talk_examples/SummaryCsv_{self.first:%Y-%m-%d}_{self.last:%Y-%m-%d}.csv'\n",
    "        )\n",
    "    \n",
    "    def run(self):\n",
    "        lines = ['date,english,french']\n",
    "        \n",
    "        for date, (english_target, french_target) in self.input().items():\n",
    "            english = english_target.open().read()\n",
    "            french = french_target.open().read()\n",
    "            lines.append(f'{date:%Y-%m-%d},{english},{french}')\n",
    "        \n",
    "        tmpfile = f'{self.output().path}.tmp'\n",
    "        with open(tmpfile, 'w') as fd:\n",
    "            fd.write('\\n'.join(lines))\n",
    "        os.rename(tmpfile, self.output().path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "![SummaryCsv(first=2019-04-01, last=2019-04-02) Dependency Graph](./SummaryCsv_dependency_graph.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO: Informed scheduler that task   SummaryCsv_2019_04_01_2019_04_02_2fcce54d60   has status   PENDING\n",
      "INFO: Informed scheduler that task   DayNameFrench_2019_04_02_2ffbc63b3b   has status   PENDING\n",
      "INFO: Informed scheduler that task   DayName_2019_04_02_2ffbc63b3b   has status   PENDING\n",
      "INFO: Informed scheduler that task   DayNameFrench_2019_04_01_5b092e82ff   has status   PENDING\n",
      "INFO: Informed scheduler that task   DayName_2019_04_01_5b092e82ff   has status   PENDING\n",
      "INFO: Done scheduling tasks\n",
      "INFO: Running Worker with 1 processes\n",
      "INFO: [pid 74] Worker Worker(salt=249099519, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) running   DayName(date=2019-04-02)\n",
      "INFO: [pid 74] Worker Worker(salt=249099519, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) done      DayName(date=2019-04-02)\n",
      "INFO: Informed scheduler that task   DayName_2019_04_02_2ffbc63b3b   has status   DONE\n",
      "INFO: [pid 74] Worker Worker(salt=249099519, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) running   DayNameFrench(date=2019-04-02)\n",
      "INFO: [pid 74] Worker Worker(salt=249099519, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) done      DayNameFrench(date=2019-04-02)\n",
      "INFO: Informed scheduler that task   DayNameFrench_2019_04_02_2ffbc63b3b   has status   DONE\n",
      "INFO: [pid 74] Worker Worker(salt=249099519, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) running   DayName(date=2019-04-01)\n",
      "INFO: [pid 74] Worker Worker(salt=249099519, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) done      DayName(date=2019-04-01)\n",
      "INFO: Informed scheduler that task   DayName_2019_04_01_5b092e82ff   has status   DONE\n",
      "INFO: [pid 74] Worker Worker(salt=249099519, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) running   DayNameFrench(date=2019-04-01)\n",
      "INFO: [pid 74] Worker Worker(salt=249099519, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) done      DayNameFrench(date=2019-04-01)\n",
      "INFO: Informed scheduler that task   DayNameFrench_2019_04_01_5b092e82ff   has status   DONE\n",
      "INFO: [pid 74] Worker Worker(salt=249099519, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) running   SummaryCsv(first=2019-04-01, last=2019-04-02)\n",
      "INFO: [pid 74] Worker Worker(salt=249099519, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) done      SummaryCsv(first=2019-04-01, last=2019-04-02)\n",
      "INFO: Informed scheduler that task   SummaryCsv_2019_04_01_2019_04_02_2fcce54d60   has status   DONE\n",
      "INFO: Worker Worker(salt=249099519, workers=1, host=DESKTOP-4FKKS4Q, username=harry, pid=74) was stopped. Shutting down Keep-Alive thread\n",
      "INFO: \n",
      "===== Luigi Execution Summary =====\n",
      "\n",
      "Scheduled 5 tasks of which:\n",
      "* 5 ran successfully:\n",
      "    - 2 DayName(date=2019-04-01,2019-04-02)\n",
      "    - 2 DayNameFrench(date=2019-04-01,2019-04-02)\n",
      "    - 1 SummaryCsv(first=2019-04-01, last=2019-04-02)\n",
      "\n",
      "This progress looks :) because there were no failed tasks or missing dependencies\n",
      "\n",
      "===== Luigi Execution Summary =====\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# luigi --module examples SummaryCsv --first 2019-04-01 --last 2019-04-02 --local-scheduler\n",
    "luigi.build(\n",
    "    [SummaryCsv(first=datetime.date(2019,4,1), last=datetime.date(2019,4,2))],\n",
    "    local_scheduler=True\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "date,english,french\r\n",
      "2019-04-01,Monday,Oh la la c'est le Monday\r\n",
      "2019-04-02,Tuesday,Oh la la c'est le Tuesday"
     ]
    }
   ],
   "source": [
    "! cat /tmp/luigi_talk_examples/SummaryCsv_2019-04-01_2019-04-02.csv"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Parallelisation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "# luigi --module examples SummaryCsv --first 2019-04-01 --last 2019-04-09 --workers 4 --local-scheduler\n",
    "luigi.build(\n",
    "    [SummaryCsv(first=datetime.date(2019,4,1), last=datetime.date(2019,4,9))],\n",
    "    local_scheduler=True,\n",
    "    workers=4\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# How can this make my data-pipelines better?\n",
    "\n",
    "- modularise-ability\n",
    "- reproduce-ability\n",
    "- extend-ability\n",
    "- port-ability\n",
    "- paralellise-ability\n",
    "\n",
    "Because:\n",
    "- Makes you be explicit about inputs/outputs\n",
    "- nice framework for interacting with inputs/outputs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Calling luigi from command-line\n",
    "\n",
    "This is the recommended way to run Luigi!\n",
    "\n",
    "`luigi --module <module_name> <TaskName> --<param-name> <param-value> --local-scheduler`\n",
    "\n",
    "e.g.\n",
    "\n",
    "`luigi --module luigi_talk_examples DayName --date 2019-04-09 --local-scheduler`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Using the central scheduler\n",
    "\n",
    "Configure: https://luigi.readthedocs.io/en/stable/central_scheduler.html\n",
    "\n",
    "-  luigid\n",
    "- localhost:8082\n",
    "\n",
    "On a machine: `luigi module my_module MyTask --my-param <param-value> --workers X`\n",
    "\n",
    "1. Machine tells central scheduler which tasks it is prepared to run and how many workers it has.\n",
    "1. Central scheduler tells machine's workers when to run those tasks.\n",
    "\n",
    "![Central scheduler scrrenshot](./luigi_central_scheduler.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Some guidance\n",
    "\n",
    "#### Debugging\n",
    "- using `breakpoint()` / `import pdb; pdb.set_trace()` in `.run()` method works with `--local-scheduler` and `--workers 1`\n",
    "\n",
    "#### Targets\n",
    "- `task.output()` does not need to be local filesystem `luigi.LocalTarget` - just an object with a `target.exists()` method.\n",
    "- e.g. `.exists()` could make a SQL query\n",
    "\n",
    "#### Making Tasks\n",
    "- keep input parameters simple\n",
    "- unique input parameters -> unique output"
   ]
  }
 ],
 "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.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}