notebooks/realtime-stream.ipynb · influxdata/influxdb-client-python

轻量养虾，开箱即用！低 Token + 稳定算力，Gitee & 模力方舟联合出品的 PocketClaw 正式开售！点击了解详情~

加入 Gitee

与超过 1400万开发者一起发现、参与优秀开源项目，私有仓库也完全免费：）

文件

克隆/下载

realtime-stream.ipynb 5.31 KB

{
 "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
}

一键复制编辑原始数据按行查看历史

提交于 2022-08-01 13:57 +08:00 . chore(deps): migrate to RxPY v4; set minimal Python version to 3.7 (#472)

马建仓 AI 助手

尝试更多

代码解读

代码找茬

代码优化

https://gitee.com/influxdata/influxdb-client-python.git

git@gitee.com:influxdata/influxdb-client-python.git

influxdata

influxdb-client-python

master

influxdata/influxdb-client-python

简介

发行版

贡献者

语言

近期动态

influxdata/influxdb-client-python .gitee-modal { width: 500px !important; }

简介

发行版

贡献者

语言

近期动态

搜索帮助

influxdata/influxdb-client-python