getting-started/frames.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Using Iguazio Frames Library for High-Performance Data Access \n",
    "iguazio `v3io_frames` is a streaming oriented multi-model (generic) data API which allow high-speed data loading and storing<br>\n",
    "frames currently support iguazio key/value, time-series, and streaming data models (called backends), additional backends will be added\n",
    "\n",
    "to use frames you first create a `client` and provide it the session and credential details, the client is used to to for 5 basic operations:\n",
    "```\n",
    "   create  - create a new key/value or time-series table or stream \n",
    "   delete  - delete the table or stream\n",
    "   read    - read data from the backend (as pandas DataFrame or dataFrame iterator)\n",
    "   write   - write one or more DataFrames into the backend\n",
    "   execute - execute a command on the backend, each backend may support multiple commands \n",
    "```   \n",
    "\n",
    "Content:\n",
    "- [Working with key/value and SQL data](kv)\n",
    "- [Working with Time-series data](#tsdb)\n",
    "- [Working with Streams](#stream)\n",
    "\n",
    "The following sections describe how to use frames, for more help and details use the internal documentation, e.g. run the following command\n",
    "```  client.read?```\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import v3io_frames as v3f\n",
    "import os\n",
    "client = v3f.Client('framesd:8081', container='users')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='kv'></a>\n",
    "## Working with key/value and SQL data\n",
    "\n",
    "### Load data from Amazon S3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "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>age</th>\n",
       "      <th>job</th>\n",
       "      <th>marital</th>\n",
       "      <th>education</th>\n",
       "      <th>default</th>\n",
       "      <th>balance</th>\n",
       "      <th>housing</th>\n",
       "      <th>loan</th>\n",
       "      <th>contact</th>\n",
       "      <th>day</th>\n",
       "      <th>month</th>\n",
       "      <th>duration</th>\n",
       "      <th>campaign</th>\n",
       "      <th>pdays</th>\n",
       "      <th>previous</th>\n",
       "      <th>poutcome</th>\n",
       "      <th>y</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>30</td>\n",
       "      <td>unemployed</td>\n",
       "      <td>married</td>\n",
       "      <td>primary</td>\n",
       "      <td>no</td>\n",
       "      <td>1787</td>\n",
       "      <td>no</td>\n",
       "      <td>no</td>\n",
       "      <td>cellular</td>\n",
       "      <td>19</td>\n",
       "      <td>oct</td>\n",
       "      <td>79</td>\n",
       "      <td>1</td>\n",
       "      <td>-1</td>\n",
       "      <td>0</td>\n",
       "      <td>unknown</td>\n",
       "      <td>no</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>33</td>\n",
       "      <td>services</td>\n",
       "      <td>married</td>\n",
       "      <td>secondary</td>\n",
       "      <td>no</td>\n",
       "      <td>4789</td>\n",
       "      <td>yes</td>\n",
       "      <td>yes</td>\n",
       "      <td>cellular</td>\n",
       "      <td>11</td>\n",
       "      <td>may</td>\n",
       "      <td>220</td>\n",
       "      <td>1</td>\n",
       "      <td>339</td>\n",
       "      <td>4</td>\n",
       "      <td>failure</td>\n",
       "      <td>no</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>35</td>\n",
       "      <td>management</td>\n",
       "      <td>single</td>\n",
       "      <td>tertiary</td>\n",
       "      <td>no</td>\n",
       "      <td>1350</td>\n",
       "      <td>yes</td>\n",
       "      <td>no</td>\n",
       "      <td>cellular</td>\n",
       "      <td>16</td>\n",
       "      <td>apr</td>\n",
       "      <td>185</td>\n",
       "      <td>1</td>\n",
       "      <td>330</td>\n",
       "      <td>1</td>\n",
       "      <td>failure</td>\n",
       "      <td>no</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>30</td>\n",
       "      <td>management</td>\n",
       "      <td>married</td>\n",
       "      <td>tertiary</td>\n",
       "      <td>no</td>\n",
       "      <td>1476</td>\n",
       "      <td>yes</td>\n",
       "      <td>yes</td>\n",
       "      <td>unknown</td>\n",
       "      <td>3</td>\n",
       "      <td>jun</td>\n",
       "      <td>199</td>\n",
       "      <td>4</td>\n",
       "      <td>-1</td>\n",
       "      <td>0</td>\n",
       "      <td>unknown</td>\n",
       "      <td>no</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>59</td>\n",
       "      <td>blue-collar</td>\n",
       "      <td>married</td>\n",
       "      <td>secondary</td>\n",
       "      <td>no</td>\n",
       "      <td>0</td>\n",
       "      <td>yes</td>\n",
       "      <td>no</td>\n",
       "      <td>unknown</td>\n",
       "      <td>5</td>\n",
       "      <td>may</td>\n",
       "      <td>226</td>\n",
       "      <td>1</td>\n",
       "      <td>-1</td>\n",
       "      <td>0</td>\n",
       "      <td>unknown</td>\n",
       "      <td>no</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   age          job  marital  education default  balance housing loan  \\\n",
       "0   30   unemployed  married    primary      no     1787      no   no   \n",
       "1   33     services  married  secondary      no     4789     yes  yes   \n",
       "2   35   management   single   tertiary      no     1350     yes   no   \n",
       "3   30   management  married   tertiary      no     1476     yes  yes   \n",
       "4   59  blue-collar  married  secondary      no        0     yes   no   \n",
       "\n",
       "    contact  day month  duration  campaign  pdays  previous poutcome   y  \n",
       "0  cellular   19   oct        79         1     -1         0  unknown  no  \n",
       "1  cellular   11   may       220         1    339         4  failure  no  \n",
       "2  cellular   16   apr       185         1    330         1  failure  no  \n",
       "3   unknown    3   jun       199         4     -1         0  unknown  no  \n",
       "4   unknown    5   may       226         1     -1         0  unknown  no  "
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# read S3 file into a data frame and show its data & metadata\n",
    "tablename = os.path.join(os.getenv('V3IO_USERNAME')+'/examples/bank')\n",
    "df = pd.read_csv('https://s3.amazonaws.com/iguazio-sample-data/bank.csv', sep=';')\n",
    "df.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Write data frames into the database using a single command\n",
    "data is streamed into the database via fast NoSQL APIs, note the backend is `kv`<br>\n",
    "the input data can be a single dataframe or a dataframe iterator (for streaming)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "out = client.write('kv', tablename, df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Read from the Database with DB side SQL\n",
    "offload data filtering, grouping, joins, etc to a scale-out high speed DB engine<br>\n",
    "Note that we're using a V3IO_USERNAME as environment variable as therefore we need to define the string for the \"From\" section<br>\n",
    "The from convention is select * from v3io.<data container>.\"path\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Done.\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "    <tr>\n",
       "        <th>loan</th>\n",
       "        <th>education</th>\n",
       "        <th>previous</th>\n",
       "        <th>housing</th>\n",
       "        <th>poutcome</th>\n",
       "        <th>duration</th>\n",
       "        <th>marital</th>\n",
       "        <th>default</th>\n",
       "        <th>balance</th>\n",
       "        <th>month</th>\n",
       "        <th>contact</th>\n",
       "        <th>campaign</th>\n",
       "        <th>y</th>\n",
       "        <th>job</th>\n",
       "        <th>day</th>\n",
       "        <th>age</th>\n",
       "        <th>pdays</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "        <td>no</td>\n",
       "        <td>secondary</td>\n",
       "        <td>0</td>\n",
       "        <td>yes</td>\n",
       "        <td>unknown</td>\n",
       "        <td>249</td>\n",
       "        <td>married</td>\n",
       "        <td>no</td>\n",
       "        <td>19317</td>\n",
       "        <td>aug</td>\n",
       "        <td>cellular</td>\n",
       "        <td>1</td>\n",
       "        <td>yes</td>\n",
       "        <td>retired</td>\n",
       "        <td>4</td>\n",
       "        <td>68</td>\n",
       "        <td>-1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "        <td>no</td>\n",
       "        <td>secondary</td>\n",
       "        <td>0</td>\n",
       "        <td>no</td>\n",
       "        <td>unknown</td>\n",
       "        <td>219</td>\n",
       "        <td>married</td>\n",
       "        <td>no</td>\n",
       "        <td>26452</td>\n",
       "        <td>jul</td>\n",
       "        <td>telephone</td>\n",
       "        <td>2</td>\n",
       "        <td>no</td>\n",
       "        <td>retired</td>\n",
       "        <td>15</td>\n",
       "        <td>75</td>\n",
       "        <td>-1</td>\n",
       "    </tr>\n",
       "</table>"
      ],
      "text/plain": [
       "[('no', 'secondary', 0, 'yes', 'unknown', 249, 'married', 'no', 19317, 'aug', 'cellular', 1, 'yes', 'retired', 4, 68, -1),\n",
       " ('no', 'secondary', 0, 'no', 'unknown', 219, 'married', 'no', 26452, 'jul', 'telephone', 2, 'no', 'retired', 15, 75, -1)]"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "table_path = os.path.join('v3io.users.\"'+os.getenv('V3IO_USERNAME')+'/examples/bank\"')\n",
    "%sql select * from $table_path where balance > 10000"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Read the data through frames API\n",
    "the frames API returns a dataframe or a dataframe iterator (a stream)<br>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "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>housing</th>\n",
       "      <th>contact</th>\n",
       "      <th>education</th>\n",
       "      <th>loan</th>\n",
       "      <th>campaign</th>\n",
       "      <th>pdays</th>\n",
       "      <th>poutcome</th>\n",
       "      <th>default</th>\n",
       "      <th>balance</th>\n",
       "      <th>duration</th>\n",
       "      <th>previous</th>\n",
       "      <th>job</th>\n",
       "      <th>marital</th>\n",
       "      <th>month</th>\n",
       "      <th>day</th>\n",
       "      <th>age</th>\n",
       "      <th>y</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>__name</th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>75</th>\n",
       "      <td>no</td>\n",
       "      <td>telephone</td>\n",
       "      <td>secondary</td>\n",
       "      <td>no</td>\n",
       "      <td>2.0</td>\n",
       "      <td>-1.0</td>\n",
       "      <td>unknown</td>\n",
       "      <td>no</td>\n",
       "      <td>26452.0</td>\n",
       "      <td>219.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>retired</td>\n",
       "      <td>married</td>\n",
       "      <td>jul</td>\n",
       "      <td>15.0</td>\n",
       "      <td>75.0</td>\n",
       "      <td>no</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "       housing    contact  education loan  campaign  pdays poutcome default  \\\n",
       "__name                                                                        \n",
       "75          no  telephone  secondary   no       2.0   -1.0  unknown      no   \n",
       "\n",
       "        balance  duration  previous      job  marital month   day   age   y  \n",
       "__name                                                                       \n",
       "75      26452.0     219.0       0.0  retired  married   jul  15.0  75.0  no  "
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df = client.read(backend='kv', table=tablename, filter=\"balance>20000\")\n",
    "df.head(8)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Read the data as a stream iterator\n",
    "to use iterator and allow cuncurent data movement and processing add `iterator=True`, you will need to iterate over the returned value or use `concat`\n",
    "iterators work with all backends (not just stream), they allow streaming when placed as an input to write functions which support iterators as input"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "        balance  campaign  marital default loan    contact   y   age  \\\n",
      "__name                                                                 \n",
      "75      26452.0       2.0  married      no   no  telephone  no  75.0   \n",
      "\n",
      "        duration  previous   day housing  pdays  education      job poutcome  \\\n",
      "__name                                                                         \n",
      "75         219.0       0.0  15.0      no   -1.0  secondary  retired  unknown   \n",
      "\n",
      "       month  \n",
      "__name        \n",
      "75       jul  \n"
     ]
    }
   ],
   "source": [
    "dfs = client.read(backend='kv', table=tablename, filter=\"balance>20000\", iterator=True)\n",
    "for df in dfs:\n",
    "    print(df.head())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Batch updates with expression\n",
    "in many cases we want to update specific column values or update a column using an expression (e.g. counter = counter + x)<br>\n",
    "when using the key/value backend it can run an expression against each of the rows (specified in the index), and use the dataframe columns as parameters<br>\n",
    "columns are specified using `{}`, e.g. specifing `expression=\"packets=packets+{pkt};bytes=bytes+{bytes};last_update={mytime}\"` will add the data in `pkt` and `bytes` column from the input dataframe to the `packets` and `bytes` columns in the row and set the `last_update` field to `mytime`. the rows are selected based on the input dataframe index"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# example: creating a new column which reflect the delta between the old `balance` column and the one provided in df (should result in 0 since df didnt change)\n",
    "out = client.write('kv', tablename, df, expression='balance_delta=balance-{balance}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Making a single row update using execute command\n",
    "The use of `condition` is optional and allow to implement safe/conditional transactions "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "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",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "Empty DataFrame\n",
       "Columns: []\n",
       "Index: []"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "client.execute('kv',tablename,'update', args={'key':'44', 'expression': 'age=44', 'condition':'balance>0'})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Delete the table\n",
    "note: in kv (NoSQL) tabels there is no need to create a table before using it"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "client.delete('kv',table=tablename)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='tsdb'></a>\n",
    "## Working with time-series data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that the tsdb table example will be created under the root of the \"users\" container"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create a time series table, rate specifies the typical ingestion rate (e.g. one sample per minute)\n",
    "client.create(backend='tsdb', table='tsdb_tab',attrs={'rate':'1/m'})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pandas.core.frame.DataFrame'>\n",
      "DatetimeIndex: 60 entries, 2019-03-14 07:05:00-05:00 to 2019-03-14 12:00:00-05:00\n",
      "Freq: 300S\n",
      "Data columns (total 3 columns):\n",
      "cpu     60 non-null float64\n",
      "mem     60 non-null float64\n",
      "disk    60 non-null float64\n",
      "dtypes: float64(3)\n",
      "memory usage: 1.9 KB\n",
      "None                                 cpu       mem      disk\n",
      "2019-03-14 07:05:00-05:00  0.300817 -0.136579  1.400577\n",
      "2019-03-14 07:10:00-05:00  0.340741 -0.485285  2.723041\n",
      "2019-03-14 07:15:00-05:00 -2.029409  0.285942  2.076960\n",
      "2019-03-14 07:20:00-05:00 -1.977885  0.096732  4.749289\n",
      "2019-03-14 07:25:00-05:00 -1.567182  0.174292  6.428432\n"
     ]
    }
   ],
   "source": [
    "# create sample time-series data\n",
    "import numpy as np\n",
    "from datetime import datetime, timedelta\n",
    "end = datetime.now().replace(minute=0, second=0, microsecond=0)\n",
    "rng = pd.date_range(end=end, periods=60, freq='300s', tz='EST')\n",
    "df = pd.DataFrame(np.random.randn(len(rng), 3), index=rng, columns=['cpu','mem','disk'])\n",
    "df = df.cumsum()\n",
    "print(df.info(), df.head())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Write to the time-series DB\n",
    "The time series DB has a time based index and additional sub indexes called labels<br>\n",
    "labels can be specified in two ways:\n",
    "* Using the `labels` parameters which will add the specified labels to each row/sample<br>\n",
    "* Using multi-index, all non time index columns are automatically converted to labels\n",
    "\n",
    "if your DataFrame doesnt contain multi-index and you wish to use specific columns as time-series labels you should convert the columns to indexes using:<br>\n",
    "```python\n",
    "    df.index.name='time'                              # in case the index column is un-named \n",
    "    df.reset_index(level=0, inplace=True)    \n",
    "    df = df.set_index(['time','symbol','exchange'])   # e.g. convert the specified columns to indexes \n",
    "```\n",
    "\n",
    "Note: you can use both (multi-index and labels) together, the labels will be the aggregation of both "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "client.write(backend='tsdb', table = 'tsdb_tab',dfs=df, labels={'node':'11'})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Read from the time-series DB\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "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></th>\n",
       "      <th>avg(mem)</th>\n",
       "      <th>max(mem)</th>\n",
       "      <th>min(mem)</th>\n",
       "      <th>avg(disk)</th>\n",
       "      <th>max(disk)</th>\n",
       "      <th>min(disk)</th>\n",
       "      <th>avg(cpu)</th>\n",
       "      <th>max(cpu)</th>\n",
       "      <th>min(cpu)</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>time</th>\n",
       "      <th>node</th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>2019-03-14 11:58:02</th>\n",
       "      <th>11</th>\n",
       "      <td>0.082436</td>\n",
       "      <td>1.478036</td>\n",
       "      <td>-1.102929</td>\n",
       "      <td>4.200892</td>\n",
       "      <td>6.428432</td>\n",
       "      <td>1.400577</td>\n",
       "      <td>-1.216931</td>\n",
       "      <td>0.340741</td>\n",
       "      <td>-2.139816</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                          avg(mem)  max(mem)  min(mem)  avg(disk)  max(disk)  \\\n",
       "time                node                                                       \n",
       "2019-03-14 11:58:02 11    0.082436  1.478036 -1.102929   4.200892   6.428432   \n",
       "\n",
       "                          min(disk)  avg(cpu)  max(cpu)  min(cpu)  \n",
       "time                node                                           \n",
       "2019-03-14 11:58:02 11     1.400577 -1.216931  0.340741 -2.139816  "
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Read Time-Series aggregates from the DB (returned as a data stream, use concat to assemble the frames)\n",
    "tsdf = client.read(backend='tsdb', query='select avg(*),max(*),min(*) from tsdb_tab', step='60m', start=\"now-7d\", end='now',multi_index=True)\n",
    "tsdf.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Delete the table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "client.delete('tsdb','tsdb_tab')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='stream'></a>\n",
    "## Working with streams\n",
    "iguazio platform support streams which have AWS Kinesis like API, they can be accessed from the notebook as if they were a structured stream (currently assume the structure is serialized through `JSON`. streams can be accessed in a bulk, or preferably using an iterator.<br>\n",
    "Streams must be first created, users can specify the number of shards and retention period when creating the stream."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "strm = os.path.join(os.getenv('V3IO_USERNAME')+'/examples/somestream')\n",
    "client.create(backend='stream', table=strm,attrs={'retention_hours':48,'shards':1})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "# write data into a stream\n",
    "def gendf():\n",
    "    end = datetime.now().replace(minute=0, second=0, microsecond=0)\n",
    "    rng = pd.date_range(end=end, periods=60, freq='300s', tz='Israel')\n",
    "    df = pd.DataFrame(np.random.randn(len(rng), 3), index=rng, columns=['cpu', 'mem', 'disk'])\n",
    "    return df\n",
    "\n",
    "client.write('stream', strm, gendf())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Reading from a stream\n",
    "the stream read operation need to specify the seek method and parameters (each seek method may have different parameters) as listed below:\n",
    "```\n",
    "   earliest   - start from the earliest point in the stream (no params)\n",
    "   latest     - start from the latest, i.e. show only new records\n",
    "   time       - start from a point in time, specify the start param e.g. start='now-1d'\n",
    "   sequence   - start from a specific sequence number, specify the sequence param e.g. sequence=45\n",
    "```\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "                 cpu      disk               index-0       mem  \\\n",
      "seq_number                                                       \n",
      "1           0.011781 -0.172618  2019-03-14T05:05:00Z  0.535711   \n",
      "2           1.580387 -0.470061  2019-03-14T05:10:00Z -0.003681   \n",
      "3           0.400377 -1.969355  2019-03-14T05:15:00Z  0.363214   \n",
      "4          -0.213012 -0.806401  2019-03-14T05:20:00Z  0.250864   \n",
      "\n",
      "                             stream_time  \n",
      "seq_number                                \n",
      "1          2019-03-14 12:58:08.343475133  \n",
      "2          2019-03-14 12:58:08.343475133  \n",
      "3          2019-03-14 12:58:08.343475133  \n",
      "4          2019-03-14 12:58:08.343475133  \n"
     ]
    }
   ],
   "source": [
    "dfs = client.read('stream', strm,seek='earliest', shard_id='0', iterator=True)\n",
    "for df in dfs:\n",
    "    print(df.head(4))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Push a single record update to a stream\n",
    "In some cases it is more conviniant to just push a buffer into a stream, for that use the execute `put` command <br>\n",
    "put accepts the `data` arg and two optional parameters (`clientinfo` for some extra info and `partition` if you want to specify the shard id)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "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",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "Empty DataFrame\n",
       "Columns: []\n",
       "Index: []"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "client.execute('stream', strm, 'put', args={'data': 'abcd', 'clientinfo': '123'})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Delete the stream"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "client.delete('stream',strm)"
   ]
  },
  {
   "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.6.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}