{
 "metadata": {
  "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.10"
  },
  "orig_nbformat": 2,
  "kernelspec": {
   "name": "python3610jvsc74a57bd0eb5e09632d6ea1cbf3eb9da7e37b7cf581db5ed13074b21cc44e159dc62acdab",
   "display_name": "Python 3.6.10 64-bit ('dataloader': conda)"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2,
 "cells": [
  {
   "source": [
    "## \\[RFC\\] How DataFrames (DF) and DataPipes (DP) work together"
   ],
   "cell_type": "markdown",
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from importlib import reload\n",
    "import torch\n",
    "reload(torch)\n",
    "from torch.utils.data import IterDataPipe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Example IterDataPipe\n",
    "class ExampleIterPipe(IterDataPipe):\n",
    "    def __init__(self, range = 20):\n",
    "        self.range = range\n",
    "    def __iter__(self):\n",
    "        for i in range(self.range):\n",
    "            yield i\n",
    "\n",
    "def get_dataframes_pipe(range = 10, dataframe_size = 7):\n",
    "    return ExampleIterPipe(range = range).map(lambda i: (i, i % 3))._to_dataframes_pipe(columns = ['i','j'], dataframe_size = dataframe_size)\n",
    "\n",
    "def get_regular_pipe(range = 10):\n",
    "    return ExampleIterPipe(range = range).map(lambda i: (i, i % 3))\n"
   ]
  },
  {
   "source": [
    "Doesn't matter how DF composed internally, iterator over DF Pipe gives single rows to user. This is similar to regular DataPipe."
   ],
   "cell_type": "markdown",
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "DataFrames Pipe\n(0, 0)\n(1, 1)\n(2, 2)\n(3, 0)\n(4, 1)\n(5, 2)\n(6, 0)\n(7, 1)\n(8, 2)\n(9, 0)\nRegular DataPipe\n(0, 0)\n(1, 1)\n(2, 2)\n(3, 0)\n(4, 1)\n(5, 2)\n(6, 0)\n(7, 1)\n(8, 2)\n(9, 0)\n"
     ]
    }
   ],
   "source": [
    "print('DataFrames Pipe')\n",
    "dp = get_dataframes_pipe()\n",
    "for i in dp:\n",
    "    print(i)\n",
    "\n",
    "print('Regular DataPipe')\n",
    "dp = get_regular_pipe()\n",
    "for i in dp:\n",
    "    print(i)"
   ]
  },
  {
   "source": [
    "You can iterate over raw DF using `raw_iterator`"
   ],
   "cell_type": "markdown",
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "   i  j\n0  0  0\n1  1  1\n2  2  2\n3  3  0\n4  4  1\n5  5  2\n6  6  0\n   i  j\n0  7  1\n1  8  2\n2  9  0\n"
     ]
    }
   ],
   "source": [
    "dp = get_dataframes_pipe()\n",
    "for i in dp.raw_iterator():\n",
    "    print(i)"
   ]
  },
  {
   "source": [
    "Operations over DF Pipe is captured"
   ],
   "cell_type": "markdown",
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "var_3 = input_var_2.i * 100\nvar_4 = var_3 + input_var_2.j\nvar_5 = var_4 - 2.7\ninput_var_2[\"y\"] = var_5\n"
     ]
    }
   ],
   "source": [
    "dp = get_dataframes_pipe(dataframe_size = 3)\n",
    "dp['y'] = dp.i * 100 + dp.j - 2.7\n",
    "print(dp.ops_str())\n"
   ]
  },
  {
   "source": [
    "Captured operations executed on `__next__` calls of constructed DataPipe"
   ],
   "cell_type": "markdown",
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "   i  j      y\n0  0  0   -2.7\n1  1  1   98.3\n2  2  2  199.3\n   i  j      y\n0  3  0  297.3\n1  4  1  398.3\n2  5  2  499.3\n   i  j      y\n0  6  0  597.3\n1  7  1  698.3\n2  8  2  799.3\n   i  j      y\n0  9  0  897.3\n"
     ]
    }
   ],
   "source": [
    "dp = get_dataframes_pipe(dataframe_size = 3)\n",
    "dp['y'] = dp.i * 100 + dp.j - 2.7\n",
    "for i in dp.raw_iterator():\n",
    "    print(i)"
   ]
  },
  {
   "source": [
    "`shuffle` of DataFramePipe effects rows in individual manner"
   ],
   "cell_type": "markdown",
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "Raw DataFrames iterator\n   i  j\n2  8  2\n2  2  2\n2  5  2\n   i  j\n1  4  1\n1  1  1\n0  3  0\n   i  j\n1  7  1\n0  9  0\n0  6  0\n   i  j\n0  0  0\nRegular DataFrames iterator\n(1, 1)\n(5, 2)\n(8, 2)\n(9, 0)\n(7, 1)\n(6, 0)\n(3, 0)\n(4, 1)\n(0, 0)\n(2, 2)\nRegular iterator\n(5, 2)\n(6, 0)\n(0, 0)\n(9, 0)\n(3, 0)\n(1, 1)\n(2, 2)\n(8, 2)\n(4, 1)\n(7, 1)\n"
     ]
    }
   ],
   "source": [
    "dp = get_dataframes_pipe(dataframe_size = 3)\n",
    "dp = dp.shuffle()\n",
    "print('Raw DataFrames iterator')\n",
    "for i in dp.raw_iterator():\n",
    "    print(i)\n",
    "\n",
    "print('Regular DataFrames iterator')\n",
    "for i in dp:\n",
    "    print(i)\n",
    "\n",
    "\n",
    "# this is similar to shuffle of regular DataPipe\n",
    "dp = get_regular_pipe()\n",
    "dp = dp.shuffle()\n",
    "print('Regular iterator')\n",
    "for i in dp:\n",
    "    print(i)"
   ]
  },
  {
   "source": [
    "You can continue mixing DF and DP operations"
   ],
   "cell_type": "markdown",
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "    i   j          y\n0 -17 -17  -197000.0\n1 -13 -16  3813000.0\n0 -11 -17  5803000.0\n    i   j          y\n2 -12 -15  4823000.0\n1 -10 -16  6813000.0\n1 -16 -16   813000.0\n    i   j          y\n0  -8 -17  8803000.0\n2  -9 -15  7823000.0\n0 -14 -17  2803000.0\n    i   j          y\n2 -15 -15  1823000.0\n"
     ]
    }
   ],
   "source": [
    "dp = get_dataframes_pipe(dataframe_size = 3)\n",
    "dp['y'] = dp.i * 100 + dp.j - 2.7\n",
    "dp = dp.shuffle()\n",
    "dp = dp - 17\n",
    "dp['y'] = dp.y * 10000\n",
    "for i in dp.raw_iterator():\n",
    "    print(i)"
   ]
  },
  {
   "source": [
    "Batching combines everything into `list` it is possible to nest `list`s. List may have any number of DataFrames as soon as total number of rows equal to batch size."
   ],
   "cell_type": "markdown",
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "Iterate over DataFrame batches\n[(6, 0),(0, 0)]\n[(4, 1),(1, 1)]\n[(2, 2),(9, 0)]\n[(3, 0),(5, 2)]\n[(7, 1),(8, 2)]\nIterate over regular batches\n[(1, 1),(4, 1)]\n[(2, 2),(3, 0)]\n[(6, 0),(7, 1)]\n[(8, 2),(0, 0)]\n[(5, 2),(9, 0)]\n"
     ]
    }
   ],
   "source": [
    "dp = get_dataframes_pipe(dataframe_size = 3)\n",
    "dp = dp.shuffle()\n",
    "dp = dp.batch(2)\n",
    "print(\"Iterate over DataFrame batches\")\n",
    "for i,v in enumerate(dp):\n",
    "    print(v)\n",
    "\n",
    "# this is similar to batching of regular DataPipe\n",
    "dp = get_regular_pipe()\n",
    "dp = dp.shuffle()\n",
    "dp = dp.batch(2)\n",
    "print(\"Iterate over regular batches\")\n",
    "for i in dp:\n",
    "    print(i)"
   ]
  },
  {
   "source": [
    "Some details about internal storage of batched DataFrames and how they are iterated"
   ],
   "cell_type": "markdown",
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "Type:  <class 'torch.utils.data.datapipes.iter.dataframes.DataChunkDF'>\n",
      "As string:  [(0, 0),(3, 0)]\n",
      "Iterated regularly:\n",
      "-- batch start --\n",
      "(0, 0)\n",
      "(3, 0)\n",
      "-- batch end --\n",
      "Iterated in inner format (for developers):\n",
      "-- df batch start --\n",
      "   i  j\n",
      "0  0  0\n",
      "0  3  0\n",
      "-- df batch end --\n",
      "Type:  <class 'torch.utils.data.datapipes.iter.dataframes.DataChunkDF'>\n",
      "As string:  [(6, 0),(1, 1)]\n",
      "Iterated regularly:\n",
      "-- batch start --\n",
      "(6, 0)\n",
      "(1, 1)\n",
      "-- batch end --\n",
      "Iterated in inner format (for developers):\n",
      "-- df batch start --\n",
      "   i  j\n",
      "0  6  0\n",
      "1  1  1\n",
      "-- df batch end --\n",
      "Type:  <class 'torch.utils.data.datapipes.iter.dataframes.DataChunkDF'>\n",
      "As string:  [(9, 0),(4, 1)]\n",
      "Iterated regularly:\n",
      "-- batch start --\n",
      "(9, 0)\n",
      "(4, 1)\n",
      "-- batch end --\n",
      "Iterated in inner format (for developers):\n",
      "-- df batch start --\n",
      "   i  j\n",
      "0  9  0\n",
      "1  4  1\n",
      "-- df batch end --\n",
      "Type:  <class 'torch.utils.data.datapipes.iter.dataframes.DataChunkDF'>\n",
      "As string:  [(5, 2),(2, 2)]\n",
      "Iterated regularly:\n",
      "-- batch start --\n",
      "(5, 2)\n",
      "(2, 2)\n",
      "-- batch end --\n",
      "Iterated in inner format (for developers):\n",
      "-- df batch start --\n",
      "   i  j\n",
      "2  5  2\n",
      "2  2  2\n",
      "-- df batch end --\n",
      "Type:  <class 'torch.utils.data.datapipes.iter.dataframes.DataChunkDF'>\n",
      "As string:  [(8, 2),(7, 1)]\n",
      "Iterated regularly:\n",
      "-- batch start --\n",
      "(8, 2)\n",
      "(7, 1)\n",
      "-- batch end --\n",
      "Iterated in inner format (for developers):\n",
      "-- df batch start --\n",
      "   i  j\n",
      "2  8  2\n",
      "1  7  1\n",
      "-- df batch end --\n"
     ]
    }
   ],
   "source": [
    "dp = get_dataframes_pipe(dataframe_size = 3)\n",
    "dp = dp.shuffle()\n",
    "dp = dp.batch(2)\n",
    "for i in dp:\n",
    "    print(\"Type: \", type(i))\n",
    "    print(\"As string: \", i)\n",
    "    print(\"Iterated regularly:\")\n",
    "    print('-- batch start --')\n",
    "    for item in i:\n",
    "        print(item)\n",
    "    print('-- batch end --')\n",
    "    print(\"Iterated in inner format (for developers):\")\n",
    "    print('-- df batch start --')\n",
    "    for item in i.raw_iterator():\n",
    "        print(item)\n",
    "    print('-- df batch end --')   "
   ]
  },
  {
   "source": [
    "`concat` should work only of DF with same schema, this code should produce an error "
   ],
   "cell_type": "markdown",
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "# TODO!\n",
    "# dp0 = get_dataframes_pipe(range = 8, dataframe_size = 4)\n",
    "# dp = get_dataframes_pipe(range = 6, dataframe_size = 3)\n",
    "# dp['y'] = dp.i * 100 + dp.j - 2.7\n",
    "# dp = dp.concat(dp0)\n",
    "# for i,v in enumerate(dp.raw_iterator()):\n",
    "#     print(v)"
   ]
  },
  {
   "source": [
    "`unbatch` of `list` with DataFrame works similarly to regular unbatch.\n",
    "Note: DataFrame sizes might change"
   ],
   "cell_type": "markdown",
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "output_type": "error",
     "ename": "AttributeError",
     "evalue": "",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mAttributeError\u001b[0m                            Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-12-fa80e9c68655>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m      4\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      5\u001b[0m \u001b[0;31m# Here is bug with unbatching which doesn't detect DF type.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 6\u001b[0;31m \u001b[0mdp\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;34m'z'\u001b[0m\u001b[0;34m]\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mdp\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0my\u001b[0m \u001b[0;34m-\u001b[0m \u001b[0;36m100\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m      7\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      8\u001b[0m \u001b[0;32mfor\u001b[0m \u001b[0mi\u001b[0m \u001b[0;32min\u001b[0m \u001b[0mdp\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mraw_iterator\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/dataset/pytorch/torch/utils/data/dataset.py\u001b[0m in \u001b[0;36m__getattr__\u001b[0;34m(self, attribute_name)\u001b[0m\n\u001b[1;32m    222\u001b[0m             \u001b[0;32mreturn\u001b[0m \u001b[0mfunction\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    223\u001b[0m         \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 224\u001b[0;31m             \u001b[0;32mraise\u001b[0m \u001b[0mAttributeError\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    225\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    226\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0m__reduce_ex__\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m*\u001b[0m\u001b[0margs\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m**\u001b[0m\u001b[0mkwargs\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mAttributeError\u001b[0m: "
     ]
    }
   ],
   "source": [
    "dp = get_dataframes_pipe(range = 18, dataframe_size = 3)\n",
    "dp['y'] = dp.i * 100 + dp.j - 2.7\n",
    "dp = dp.batch(5).batch(3).batch(1).unbatch(unbatch_level = 3)\n",
    "\n",
    "# Here is bug with unbatching which doesn't detect DF type.\n",
    "dp['z'] = dp.y - 100\n",
    "\n",
    "for i in dp.raw_iterator():\n",
    "    print(i)"
   ]
  },
  {
   "source": [
    "`map` applied to individual rows, `nesting_level` argument used to penetrate batching"
   ],
   "cell_type": "markdown",
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "Iterate over DataFrame batches\n[(1111000, 1111000),(1112000, 1112000),(1113000, 1113000),(1114000, 1111000),(1115000, 1112000)]\n[(1116000, 1113000),(1117000, 1111000),(1118000, 1112000),(1119000, 1113000),(1120000, 1111000)]\nIterate over regular batches\n[(1111000, 0),(1112000, 1),(1113000, 2),(1114000, 0),(1115000, 1)]\n[(1116000, 2),(1117000, 0),(1118000, 1),(1119000, 2),(1120000, 0)]\n"
     ]
    }
   ],
   "source": [
    "dp = get_dataframes_pipe(range = 10, dataframe_size = 3)\n",
    "dp = dp.map(lambda x: x + 1111)\n",
    "dp = dp.batch(5).map(lambda x: x * 1000, nesting_level = 1)\n",
    "print(\"Iterate over DataFrame batches\")\n",
    "for i in dp:\n",
    "    print(i)\n",
    "\n",
    "# Similarly works on row level for classic DataPipe elements\n",
    "dp = get_regular_pipe(range = 10)\n",
    "dp = dp.map(lambda x: (x[0] + 1111, x[1]))\n",
    "dp = dp.batch(5).map(lambda x: (x[0] * 1000, x[1]), nesting_level = 1)\n",
    "print(\"Iterate over regular batches\")\n",
    "for i in dp:\n",
    "    print(i)\n",
    "\n"
   ]
  },
  {
   "source": [
    "`filter` applied to individual rows, `nesting_level` argument used to penetrate batching"
   ],
   "cell_type": "markdown",
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "text": [
      "Iterate over DataFrame batches\n[(6, 0),(7, 1),(8, 2),(9, 0),(10, 1)]\n[(11, 2),(12, 0)]\nIterate over regular batches\n[(6, 0),(7, 1),(8, 2),(9, 0),(10, 1)]\n[(11, 2),(12, 0)]\n"
     ]
    }
   ],
   "source": [
    "dp = get_dataframes_pipe(range = 30, dataframe_size = 3)\n",
    "dp = dp.filter(lambda x: x.i > 5)\n",
    "dp = dp.batch(5).filter(lambda x: x.i < 13, nesting_level = 1)\n",
    "print(\"Iterate over DataFrame batches\")\n",
    "for i in dp:\n",
    "    print(i)\n",
    "\n",
    "# Similarly works on row level for classic DataPipe elements\n",
    "dp = get_regular_pipe(range = 30)\n",
    "dp = dp.filter(lambda x: x[0] > 5)\n",
    "dp = dp.batch(5).filter(lambda x: x[0] < 13, nesting_level = 1)\n",
    "print(\"Iterate over regular batches\")\n",
    "for i in dp:\n",
    "    print(i)"
   ]
  }
 ]
}