docs/example-functional.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "import d6tflow\n",
    "import pandas as pd\n",
    "\n",
    "from d6tflow.functional import Workflow\n",
    "flow = Workflow()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Define workflow"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "@flow.task(d6tflow.tasks.TaskCache)\n",
    "def get_data1(task):\n",
    "    df = pd.DataFrame({'a':range(3)})\n",
    "    task.save(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "@flow.task(d6tflow.tasks.TaskCache)\n",
    "def get_data2(task):\n",
    "    df = pd.DataFrame({'a':range(3)})\n",
    "    task.save(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "@flow.task(d6tflow.tasks.TaskPqPandas)\n",
    "@flow.requires({\"input1\":get_data1, \"input2\":get_data2})\n",
    "def use_data(task):\n",
    "    data = task.inputLoad()\n",
    "    df1 = data['input1']\n",
    "    df2 = data['input2']\n",
    "    df3 = df1.join(df2, lsuffix='1', rsuffix='2')\n",
    "    df3['b']=df3['a1']*task.multiplier # use task parameter\n",
    "    task.save(df3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Execute workflow"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "flow.add_global_params(multiplier=d6tflow.IntParameter(default=0))\n",
    "use_params = {'multiplier':4}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "\n ===== Luigi Execution Preview ===== \n\n\n└─--[use_data-{'multiplier': '4'} (\u001b[92mCOMPLETE\u001b[0m)]\n   |--[get_data1- (\u001b[92mCOMPLETE\u001b[0m)]\n   └─--[get_data2- (\u001b[92mCOMPLETE\u001b[0m)]\n\n ===== Luigi Execution Preview ===== \n\n"
     ]
    }
   ],
   "source": [
    "flow.preview(use_data, params=use_params)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "\n===== Luigi Execution Summary =====\n\nScheduled 3 tasks of which:\n* 3 ran successfully:\n    - 1 get_data1(multiplier=4)\n    - 1 get_data2(multiplier=4)\n    - 1 use_data(multiplier=4)\n\nThis progress looks :) because there were no failed tasks or missing dependencies\n\n===== Luigi Execution Summary =====\n\n"
     ]
    }
   ],
   "source": [
    "flow.run(use_data, params=use_params, forced_all_upstream=True, confirm=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "   a1  a2  b\n",
       "0   0   0  0\n",
       "1   1   1  4\n",
       "2   2   2  8"
      ],
      "text/html": "<div>\n<style scoped>\n    .dataframe tbody tr th:only-of-type {\n        vertical-align: middle;\n    }\n\n    .dataframe tbody tr th {\n        vertical-align: top;\n    }\n\n    .dataframe thead th {\n        text-align: right;\n    }\n</style>\n<table border=\"1\" class=\"dataframe\">\n  <thead>\n    <tr style=\"text-align: right;\">\n      <th></th>\n      <th>a1</th>\n      <th>a2</th>\n      <th>b</th>\n    </tr>\n  </thead>\n  <tbody>\n    <tr>\n      <th>0</th>\n      <td>0</td>\n      <td>0</td>\n      <td>0</td>\n    </tr>\n    <tr>\n      <th>1</th>\n      <td>1</td>\n      <td>1</td>\n      <td>4</td>\n    </tr>\n    <tr>\n      <th>2</th>\n      <td>2</td>\n      <td>2</td>\n      <td>8</td>\n    </tr>\n  </tbody>\n</table>\n</div>"
     },
     "metadata": {},
     "execution_count": 17
    }
   ],
   "source": [
    "flow.outputLoad(use_data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "output_type": "execute_result",
     "data": {
      "text/plain": [
       "{'use_data':    a1  a2  b\n",
       " 0   0   0  0\n",
       " 1   1   1  4\n",
       " 2   2   2  8,\n",
       " 'get_data1':    a\n",
       " 0  0\n",
       " 1  1\n",
       " 2  2,\n",
       " 'get_data2':    a\n",
       " 0  0\n",
       " 1  1\n",
       " 2  2}"
      ]
     },
     "metadata": {},
     "execution_count": 19
    }
   ],
   "source": [
    "flow.outputLoadAll(use_data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "name": "python382jvsc74a57bd08f7bfc9fa91ad8da7cdcb954e1704bbe8312ef1d6773b7db4231479490ec9878",
   "display_name": "Python 3.8.2 64-bit"
  },
  "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.8.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}