stream_to_parquet/stream_to_parquet.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Stream to Parquet"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Part of the [network operations](https://github.com/mlrun/demos/tree/0.7.x/network-operations) demo pipeline, this function listens to a labeld stream and writes it as parquet files.<br>\n",
    "This function also deploys the function [virtual_drift](https://github.com/mlrun/functions/tree/master/virtual_drift) from the hub, which computes drift magnitude metrics between base dataset t and dataset u,<br>\n",
    "in our case (as well as in the demo) - base dataset (the one that the model trained on) and the dataset the model predicted.<br>\n",
    "virtual_drift writes the output to TSDB."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **Steps**\n",
    "\n",
    "1. [Data exploration](#Data-exploration)\n",
    "2. [Creating the labeled stream](#Creating-the-labeled-stream)\n",
    "3. [Importing the function](#Importing-the-function)\n",
    "4. [Running the functioh remotely](#Running-the-function-remotely)\n",
    "5. [Testing the function](#Testing-the-function)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **Data exploration**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In order to know about the performance of a drift detector by measuring the different detection metrics, we need to know beforehand where a real drift occurs.<br>\n",
    "This is only possible with synthetic datasets.<br> The scikit-multiflow framework allows generating several kinds of synthetic data to simulate the occurrence of drifts.<br>\n",
    "[Harvard dataverse](https://dataverse.harvard.edu) provides futher explanations on the [used dataset](https://dataverse.harvard.edu/dataset.xhtml?persistentId=doi:10.7910/DVN/5OWRGB) along with different kinds of drifted datasets.<br>\n",
    "mixed_0101_abrupto has 4 concepts and 3 drifts at time steps 10000, 20000, and 30000.<br>\n",
    "Our dataset will be train-test-splitted, the train part (first 5000 examples) is used to train the model (that is generated easly using [sklearn_classifer](https://github.com/mlrun/functions/blob/master/sklearn_classifier/sklearn_classifier.ipynb)). <br>\n",
    "The test part (which is already predicted by the model) will be pushed to the input stream in order to detect drifts."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "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>X1</th>\n",
       "      <th>X2</th>\n",
       "      <th>X3</th>\n",
       "      <th>X4</th>\n",
       "      <th>class</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>0.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.460101</td>\n",
       "      <td>0.592744</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.588788</td>\n",
       "      <td>0.574984</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.401641</td>\n",
       "      <td>0.679325</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.306076</td>\n",
       "      <td>0.182108</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.962847</td>\n",
       "      <td>0.579245</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "    X1   X2        X3        X4  class\n",
       "0  0.0  1.0  0.460101  0.592744    1.0\n",
       "1  1.0  1.0  0.588788  0.574984    0.0\n",
       "2  0.0  0.0  0.401641  0.679325    1.0\n",
       "3  1.0  1.0  0.306076  0.182108    0.0\n",
       "4  0.0  0.0  0.962847  0.579245    1.0"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import pandas as pd\n",
    "data_path = 'https://s3.wasabisys.com/iguazio/data/function-marketplace-data/concept_drift/mixed_0101_abrupto.csv'\n",
    "base_dataset = 'https://s3.wasabisys.com/iguazio/data/function-marketplace-data/concept_drift/predicted_abrupto_train.csv'\n",
    "# The predicted test data is pushed to the stream\n",
    "predicted_test_data_path = 'https://s3.wasabisys.com/iguazio/data/function-marketplace-data/concept_drift/predicted_abrupto_test.csv'\n",
    "# You can find the model used here\n",
    "models_path = 'https://s3.wasabisys.com/iguazio/models/function-marketplace-models/concept_drift/concept_drift_random_forest.pkl'\n",
    "original_data = pd.read_csv(data_path)\n",
    "original_data.head()"
   ]
  },
  {
   "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>X1</th>\n",
       "      <th>X2</th>\n",
       "      <th>X3</th>\n",
       "      <th>X4</th>\n",
       "      <th>class</th>\n",
       "      <th>predicted_col</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>34995</th>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.010106</td>\n",
       "      <td>0.647269</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>34996</th>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.293651</td>\n",
       "      <td>0.737291</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>34997</th>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.848546</td>\n",
       "      <td>0.552337</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>34998</th>\n",
       "      <td>1.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.614754</td>\n",
       "      <td>0.859896</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>34999</th>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.265306</td>\n",
       "      <td>0.843716</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1.0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "        X1   X2        X3        X4  class  predicted_col\n",
       "34995  0.0  0.0  0.010106  0.647269    0.0            1.0\n",
       "34996  1.0  1.0  0.293651  0.737291    1.0            0.0\n",
       "34997  0.0  0.0  0.848546  0.552337    0.0            1.0\n",
       "34998  1.0  1.0  0.614754  0.859896    1.0            0.0\n",
       "34999  1.0  0.0  0.265306  0.843716    0.0            1.0"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "predicted_test = pd.read_csv(predicted_test_data_path)\n",
    "predicted_test.tail()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **Creating the labeled stream**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os \n",
    "\n",
    "container = os.path.join('/',os.environ['V3IO_HOME'].split('/')[0])\n",
    "user = os.environ[\"V3IO_USERNAME\"]\n",
    "rel_path = os.getcwd()[6:] + '/artifacts'\n",
    "\n",
    "base_input_stream = os.path.join(user,rel_path) + \"/inputs_stream\"\n",
    "base_output_stream = os.path.join(user,rel_path) + \"/output_stream\"\n",
    "input_stream = os.path.join(container,base_input_stream)\n",
    "tsdb_path = os.path.join(user,rel_path) + \"/output_tsdb\"\n",
    "\n",
    "stream_consumer_group = 's2p'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "import v3io.dataplane\n",
    "\n",
    "client = v3io.dataplane.Client()\n",
    "response = client.stream.create(container = container,\n",
    "                                stream_path=base_input_stream,\n",
    "                                shard_count=1,\n",
    "                                raise_for_status = v3io.dataplane.RaiseForStatus.never)\n",
    "response.raise_for_status([409, 204])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **Importing the function**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "> 2021-10-26 14:37:45,224 [info] created and saved project function-marketplace\n"
     ]
    }
   ],
   "source": [
    "import mlrun\n",
    "\n",
    "# Importing the function\n",
    "mlrun.set_environment(project='function-marketplace')\n",
    "\n",
    "fn = mlrun.import_function(\"hub://stream_to_parquet:development\")\n",
    "fn.apply(mlrun.auto_mount())\n",
    "\n",
    "fn.add_v3io_stream_trigger(stream_path=input_stream, name='stream', group=stream_consumer_group)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **Running the function remotely**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "> 2021-10-26 14:37:45,513 [info] Starting remote function deploy\n",
      "2021-10-26 14:37:45  (info) Deploying function\n",
      "2021-10-26 14:37:45  (info) Building\n",
      "2021-10-26 14:37:45  (info) Staging files and preparing base images\n",
      "2021-10-26 14:37:45  (info) Building processor image\n",
      "2021-10-26 14:37:47  (info) Build complete\n",
      "2021-10-26 14:37:55  (info) Function deploy complete\n",
      "> 2021-10-26 14:37:55,689 [info] successfully deployed function: {'internal_invocation_urls': ['nuclio-function-marketplace-stream-to-parquet.default-tenant.svc.cluster.local:8080'], 'external_invocation_urls': ['default-tenant.app.dev39.lab.iguazeng.com:31445']}\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "'http://default-tenant.app.dev39.lab.iguazeng.com:31445'"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import json\n",
    "fn.set_envs({'window': 200,\n",
    "             'save_to': os.path.join(os.path.join('/User',rel_path), 'inference_pq'),\n",
    "             'prediction_col': 'predicted_col',\n",
    "             'label_col': 'class',\n",
    "             'base_dataset': base_dataset,\n",
    "             'results_tsdb_container': container[1:],\n",
    "             'results_tsdb_table': tsdb_path,\n",
    "             'mount_path': os.path.join(container,user),\n",
    "             'mount_remote': container,\n",
    "             'artifact_path': os.path.join('/User',rel_path)})\n",
    "\n",
    "fn.deploy()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **Testing the function**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'data': '{\"request\": {\"instances\": [{\"X1\": 0.0, \"X2\": 0.0, \"X3\": 0.0634475073, \"X4\": 0.4136568818, \"class\": 1.0, \"predicted_col\": 1.0}]}, \"resp\": [1], \"when\": \"2021-10-26 14:37:55.864974\", \"model\": \"sklearn.ensemble.RandomForestClassifier\"}'}"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import json\n",
    "import datetime\n",
    "\n",
    "# Reshaping the data to V3IOStream format.\n",
    "def restructure_stream_event(context, event):\n",
    "    instances = [dict()]\n",
    "    for key in predicted_test.keys():\n",
    "        if key not in ['when', 'model', 'worker', 'hostname', 'predicted_col']:\n",
    "            instances[0].update({key: event.pop(key)})\n",
    "    instances[0].update({key: event.get(key)})      \n",
    "    event['request'] = {'instances': instances}\n",
    "    event['resp'] = [int(event.pop('predicted_col'))]\n",
    "    event['when'] = datetime.datetime.strftime(datetime.datetime.now(), format=\"%Y-%m-%d %H:%M:%S.%f\")\n",
    "    event['model'] = 'sklearn.ensemble.RandomForestClassifier'\n",
    "    return event\n",
    "    \n",
    "    \n",
    "records = json.loads(predicted_test.to_json(orient='records'))\n",
    "records = [{'data': json.dumps(restructure_stream_event(context, record))} for record in records]\n",
    "\n",
    "# showing first record\n",
    "records[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Pushing some data to the input stream\n",
    "step = 500\n",
    "for i in range(0,20000,step):\n",
    "    response = client.stream.put_records(container=container,\n",
    "                                              stream_path=base_input_stream, \n",
    "                                              records=records[i:i+step])"
   ]
  },
  {
   "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>class_shift_helinger</th>\n",
       "      <th>class_shift_kld</th>\n",
       "      <th>class_shift_tvd</th>\n",
       "      <th>prior_helinger</th>\n",
       "      <th>prior_kld</th>\n",
       "      <th>prior_tvd</th>\n",
       "      <th>stream</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>time</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>2021-10-26 14:38:08.027000+00:00</th>\n",
       "      <td>0.001759</td>\n",
       "      <td>0.000025</td>\n",
       "      <td>0.002488</td>\n",
       "      <td>1.0</td>\n",
       "      <td>10.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>some_stream</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2021-10-26 14:38:08.699000+00:00</th>\n",
       "      <td>0.001759</td>\n",
       "      <td>0.000025</td>\n",
       "      <td>0.002488</td>\n",
       "      <td>1.0</td>\n",
       "      <td>10.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>some_stream</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2021-10-26 14:38:09.599000+00:00</th>\n",
       "      <td>0.001759</td>\n",
       "      <td>0.000025</td>\n",
       "      <td>0.002488</td>\n",
       "      <td>1.0</td>\n",
       "      <td>10.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>some_stream</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2021-10-26 14:38:10.759000+00:00</th>\n",
       "      <td>0.001759</td>\n",
       "      <td>0.000025</td>\n",
       "      <td>0.002488</td>\n",
       "      <td>1.0</td>\n",
       "      <td>10.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>some_stream</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2021-10-26 14:38:11.561000+00:00</th>\n",
       "      <td>0.001759</td>\n",
       "      <td>0.000025</td>\n",
       "      <td>0.002488</td>\n",
       "      <td>1.0</td>\n",
       "      <td>10.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>some_stream</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>...</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2021-10-26 14:39:42.037000+00:00</th>\n",
       "      <td>0.001759</td>\n",
       "      <td>0.000025</td>\n",
       "      <td>0.002488</td>\n",
       "      <td>1.0</td>\n",
       "      <td>10.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>some_stream</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2021-10-26 14:39:42.191000+00:00</th>\n",
       "      <td>0.001759</td>\n",
       "      <td>0.000025</td>\n",
       "      <td>0.002488</td>\n",
       "      <td>1.0</td>\n",
       "      <td>10.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>some_stream</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2021-10-26 14:39:42.586000+00:00</th>\n",
       "      <td>0.001759</td>\n",
       "      <td>0.000025</td>\n",
       "      <td>0.002488</td>\n",
       "      <td>1.0</td>\n",
       "      <td>10.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>some_stream</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2021-10-26 14:39:42.816000+00:00</th>\n",
       "      <td>0.001759</td>\n",
       "      <td>0.000025</td>\n",
       "      <td>0.002488</td>\n",
       "      <td>1.0</td>\n",
       "      <td>10.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>some_stream</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2021-10-26 14:39:49.180000+00:00</th>\n",
       "      <td>0.001759</td>\n",
       "      <td>0.000025</td>\n",
       "      <td>0.002488</td>\n",
       "      <td>1.0</td>\n",
       "      <td>10.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>some_stream</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>99 rows × 7 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "                                  class_shift_helinger  class_shift_kld  \\\n",
       "time                                                                      \n",
       "2021-10-26 14:38:08.027000+00:00              0.001759         0.000025   \n",
       "2021-10-26 14:38:08.699000+00:00              0.001759         0.000025   \n",
       "2021-10-26 14:38:09.599000+00:00              0.001759         0.000025   \n",
       "2021-10-26 14:38:10.759000+00:00              0.001759         0.000025   \n",
       "2021-10-26 14:38:11.561000+00:00              0.001759         0.000025   \n",
       "...                                                ...              ...   \n",
       "2021-10-26 14:39:42.037000+00:00              0.001759         0.000025   \n",
       "2021-10-26 14:39:42.191000+00:00              0.001759         0.000025   \n",
       "2021-10-26 14:39:42.586000+00:00              0.001759         0.000025   \n",
       "2021-10-26 14:39:42.816000+00:00              0.001759         0.000025   \n",
       "2021-10-26 14:39:49.180000+00:00              0.001759         0.000025   \n",
       "\n",
       "                                  class_shift_tvd  prior_helinger  prior_kld  \\\n",
       "time                                                                           \n",
       "2021-10-26 14:38:08.027000+00:00         0.002488             1.0       10.0   \n",
       "2021-10-26 14:38:08.699000+00:00         0.002488             1.0       10.0   \n",
       "2021-10-26 14:38:09.599000+00:00         0.002488             1.0       10.0   \n",
       "2021-10-26 14:38:10.759000+00:00         0.002488             1.0       10.0   \n",
       "2021-10-26 14:38:11.561000+00:00         0.002488             1.0       10.0   \n",
       "...                                           ...             ...        ...   \n",
       "2021-10-26 14:39:42.037000+00:00         0.002488             1.0       10.0   \n",
       "2021-10-26 14:39:42.191000+00:00         0.002488             1.0       10.0   \n",
       "2021-10-26 14:39:42.586000+00:00         0.002488             1.0       10.0   \n",
       "2021-10-26 14:39:42.816000+00:00         0.002488             1.0       10.0   \n",
       "2021-10-26 14:39:49.180000+00:00         0.002488             1.0       10.0   \n",
       "\n",
       "                                  prior_tvd       stream  \n",
       "time                                                      \n",
       "2021-10-26 14:38:08.027000+00:00        1.0  some_stream  \n",
       "2021-10-26 14:38:08.699000+00:00        1.0  some_stream  \n",
       "2021-10-26 14:38:09.599000+00:00        1.0  some_stream  \n",
       "2021-10-26 14:38:10.759000+00:00        1.0  some_stream  \n",
       "2021-10-26 14:38:11.561000+00:00        1.0  some_stream  \n",
       "...                                     ...          ...  \n",
       "2021-10-26 14:39:42.037000+00:00        1.0  some_stream  \n",
       "2021-10-26 14:39:42.191000+00:00        1.0  some_stream  \n",
       "2021-10-26 14:39:42.586000+00:00        1.0  some_stream  \n",
       "2021-10-26 14:39:42.816000+00:00        1.0  some_stream  \n",
       "2021-10-26 14:39:49.180000+00:00        1.0  some_stream  \n",
       "\n",
       "[99 rows x 7 columns]"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Reading from TSDB\n",
    "import v3io_frames as v3f\n",
    "\n",
    "v3f_client = v3f.Client(os.environ[\"V3IO_FRAMESD\"],container=container[1:])\n",
    "v3f_client.read(backend='tsdb',table=tsdb_path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[Back to the top](#Stream-to-Parquet)"
   ]
  }
 ],
 "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": 4
}