feat: Add live streaming example

Author: bednar

notebooks/realtime-stream.ipynb

@@ -57,9 +57,7 @@
     "        .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'|> filter(fn: (r) => r._measurement == \"cpu\") '\n",
-    "                                    f'|> filter(fn: (r) => r._field == \"usage_user\") '\n",
-    "                                    f'|> filter(fn: (r) => r.cpu == \"cpu-total\")')) \\\n",
+    "                                    f'|> filter(fn: (r) => (r._measurement == \"cpu\") or (r._measurement == \"mem\")) ')) \\\n",
     "        .pipe(ops.map(lambda query: client.query_api().query_stream(query))) \\\n",
     "        .pipe(ops.flat_map(lambda records: rx.from_iterable(records))) \\\n",
     "        .subscribe(observer=lambda record: sink.emit(record), on_error=lambda error: print(error))\n",
@@ -84,12 +82,28 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "example = pd.DataFrame({'value': []}, columns=['value'])\n",
+    "cpu_example = pd.DataFrame({'value': []}, columns=['value'])\n",
     "\n",
     "cpu_sink = sink\\\n",
     "    .filter(lambda record: (record[\"_measurement\"] == \"cpu\") & (record[\"_field\"] == \"usage_user\"))\\\n",
     "    .map(lambda record: pd.DataFrame({'value': [record[\"_value\"]]}, columns=['value'], index=[record[\"_time\"]]))\n",
-    "cpu = DataFrame(cpu_sink, example=example)"
+    "cpu = DataFrame(cpu_sink, example=cpu_example)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "mem_example = pd.DataFrame({'field': [], 'value': []}, columns=['field', 'value'])\n",
+    "\n",
+    "mem_sink = sink \\\n",
+    "    .filter(lambda record: record[\"_measurement\"] == \"mem\") \\\n",
+    "    .filter(lambda record: record[\"_field\"] in [\"total\", \"used\", \"free\", \"available\"]) \\\n",
+    "    .map(lambda record: pd.DataFrame({'field': record[\"_field\"], 'value': record[\"_value\"]},\n",
+    "                                     columns=['field', 'value'], index=[record[\"_time\"], record[\"_field\"]]))\n",
+    "mem = DataFrame(mem_sink, example=mem_example)"
    ]
   },
   {
@@ -119,6 +133,15 @@
     "cpu.hvplot(width=700, backlog=50, title='CPU % usage', xlabel='Time', ylabel='%', xformatter=formatter)"
    ]
   },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    " mem.groupby('field').sum().hvplot.bar()"
+   ]
+  },
   {
    "cell_type": "code",
    "execution_count": null,