notebooks/realtime-stream.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Example InfluxDB Jupyter notebook - stream data\n",
    "\n",
    "This example demonstrates how to query data from InfluxDB 2.0 using Flux and display results in real time.\n",
    "\n",
    "Prerequisites:\n",
    "1. Start InfluxDB: `./scripts/influxdb-restart.sh`\n",
    "2. Start Telegraf: `telegraf  -config ./notebooks/telegraf.conf`\n",
    "3. Install the following dependencies: `rx`, `pandas`, `streamz`, `hvplot`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import a Client\n",
    "\n",
    "import os\n",
    "import sys\n",
    "\n",
    "sys.path.insert(0, os.path.abspath('../'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from datetime import timedelta\n",
    "from typing import List\n",
    "\n",
    "import hvplot.streamz\n",
    "import pandas as pd\n",
    "import reactivex as rx\n",
    "from reactivex import operators as ops\n",
    "\n",
    "from streamz.dataframe import Random, DataFrame\n",
    "from streamz import Stream\n",
    "from influxdb_client import InfluxDBClient"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def source_data(auto_refresh: int, query: str, sink: Stream):\n",
    "    rx \\\n",
    "        .interval(period=timedelta(seconds=auto_refresh)) \\\n",
    "        .pipe(ops.map(lambda start: f'from(bucket: \"my-bucket\") '\n",
    "                                    f'|> range(start: -{auto_refresh}s, stop: now()) '\n",
    "                                    f'{query}')) \\\n",
    "        .pipe(ops.map(lambda query: client.query_api().query_data_frame(query, data_frame_index=['_time']))) \\\n",
    "        .pipe(ops.map(lambda data_frame: data_frame.drop(columns=['result', 'table']))) \\\n",
    "        .subscribe(on_next=lambda data_frame: sink.emit(data_frame), on_error=lambda error: print(error))\n",
    "    pass"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client = InfluxDBClient(url='http://localhost:8086', token='my-token', org='my-org')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cpu_query = '|> filter(fn: (r) => r._measurement == \"cpu\") ' \\\n",
    "            '|> filter(fn: (r) => r._field == \"usage_user\") ' \\\n",
    "            '|> filter(fn: (r) => r.cpu == \"cpu-total\") ' \\\n",
    "            '|> keep(columns: [\"_time\", \"_value\"])'\n",
    "\n",
    "\n",
    "cpu_sink = Stream()\n",
    "cpu_example = pd.DataFrame({'_value': []}, columns=['_value'])\n",
    "cpu_df = DataFrame(cpu_sink, example=cpu_example)\n",
    "\n",
    "source_data(auto_refresh=5, sink=cpu_sink, query=cpu_query)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "mem_query = '|> filter(fn: (r) => r._measurement == \"mem\") ' \\\n",
    "            '|> filter(fn: (r) => r._field == \"available\" or r._field == \"free\" or r._field == \"total\" or r._field == \"used\") ' \\\n",
    "            '|> map(fn: (r) => ({ r with _value: r._value / 1024 / 1024 }))' \\\n",
    "            '|> pivot(rowKey:[\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\")' \\\n",
    "            '|> keep(columns: [\"_time\", \"used\", \"total\", \"free\", \"available\"])'\n",
    "\n",
    "mem_sink = Stream()\n",
    "mem_example = pd.DataFrame({'used': [], 'total': [], 'free': [], 'available': []}, columns=['available', 'free', 'total', 'used'])\n",
    "mem_df = DataFrame(mem_sink, example=mem_example)\n",
    "\n",
    "source_data(auto_refresh=5, sink=mem_sink, query=mem_query)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "from bokeh.models.formatters import DatetimeTickFormatter\n",
    "\n",
    "# Time formatter\n",
    "formatter = DatetimeTickFormatter(\n",
    "    microseconds = [\"%H:%M:%S\"],\n",
    "    milliseconds = [\"%H:%M:%S\"],\n",
    "    seconds = [\"%H:%M:%S\"],\n",
    "    minsec = [\"%H:%M:%S\"],\n",
    "    minutes = [\"%H:%M:%S\"],\n",
    "    hourmin = [\"%H:%M:%S\"],\n",
    "    hours=[\"%H:%M:%S\"],\n",
    "    days=[\"%H:%M:%S\"],\n",
    "    months=[\"%H:%M:%S\"],\n",
    "    years=[\"%H:%M:%S\"],\n",
    ")\n",
    "\n",
    "cpu_df.hvplot(width=450, backlog=50, title='CPU % usage', xlabel='Time', ylabel='%', xformatter=formatter) +\\\n",
    "mem_df.hvplot.line(width=450, backlog=50, title='Memory', xlabel='Time', ylabel='MiB', xformatter=formatter, legend='top_left')"
   ]
  },
  {
   "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.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}