影刀RPA 时序数据库InfluxDB：监控指标自动采集与看板

影刀RPA 时序数据库InfluxDB监控指标自动采集与看板作者林焱 | 适用人群影刀RPA新手 | 难度★★★☆☆一、什么情况用这个运维和数据分析的同学经常面对这种需求每天记录服务器的CPU使用率、网站的访问量、生产线的产量数据然后生成趋势图。存Excel吧一个月的数据就几十万行打开都费劲。存MySQL吧按时间范围查询还行但做聚合分析每小时平均、每日峰值SQL写起来很痛苦。时序数据库InfluxDB专为这种场景设计。用影刀定时采集指标 → 写入InfluxDB → 用Grafana出图一条龙搞定。实际场景服务器资源监控、业务指标追踪、IoT传感器数据存储、API调用量统计。这篇文章能解决的问题影刀Python节点连接InfluxDB定时采集指标数据并写入查询历史数据做聚合分析与Grafana配合搭建实时监控看板二、怎么做2.1 环境准备影刀安装目录\python\python.exe-mpipinstallinfluxdb-client确保有InfluxDB 2.x实例可用本地或云端。拼多多店群自动化报活动上架2.2 影刀主流程┌──────────────────────────────────────────────┐ │ ① 【执行Python代码】 │ │ 采集系统指标CPU/内存/磁盘 │ │ ↓ │ │ ② 【执行Python代码】 │ │ 将指标写入InfluxDB │ │ ↓ │ │ ③ 【执行Python代码】 │ │ 查询历史数据计算均值/峰值 │ │ ↓ │ │ ④ 【条件判断】 │ │ 判断是否超阈值 → 是则告警 │ │ ↓ │ │ ⑤ 【发送企业微信消息】 │ │ 超标告警通知 │ └──────────────────────────────────────────────┘2.3 写入指标数据步骤①②采集并写入importpsutilfrominfluxdb_clientimportInfluxDBClient,Pointfrominfluxdb_client.client.write_apiimportSYNCHRONOUSfromdatetimeimportdatetime# InfluxDB连接配置 INFLUX_URLhttp://localhost:8086INFLUX_TOKENyour-api-tokenINFLUX_ORGmy-orgINFLUX_BUCKETserver-metrics# 创建客户端 clientInfluxDBClient(urlINFLUX_URL,tokenINFLUX_TOKEN,orgINFLUX_ORG)write_apiclient.write_api(write_optionsSYNCHRONOUS)# 采集系统指标 cpu_percentpsutil.cpu_percent(interval1)memorypsutil.virtual_memory()diskpsutil.disk_usage(/)# 构造数据点 server_nameweb-server-01# 服务器标识points[# CPU使用率Point(cpu)\.tag(host,server_name)\.field(usage_percent,cpu_percent)\.time(datetime.utcnow()),# 内存Point(memory)\.tag(host,server_name)\.field(total_gb,round(memory.total/(1024**3),2))\.field(used_gb,round(memory.used/(1024**3),2))\.field(percent,memory.percent)\.time(datetime.utcnow()),# 磁盘Point(disk)\.tag(host,server_name)\.tag(mount,/)\.field(total_gb,round(disk.total/(1024**3),2))\.field(used_gb,round(disk.used/(1024**3),2))\.field(percent,disk.percent)\.time(datetime.utcnow()),]# 写入 write_api.write(bucketINFLUX_BUCKET,orgINFLUX_ORG,recordpoints)print(f写入成功CPU{cpu_percent}%, 内存{memory.percent}%, 磁盘{disk.percent}%)client.close()关键理解InfluxDB的数据模型是measurement(表) tags(标签/索引) fields(数值字段) timestamp(时间戳)。tags用于分组过滤fields用于存储实际数值。2.4 从API采集业务指标如果是采集业务数据比如网站PV/UV而非系统指标importrequestsfrominfluxdb_clientimportInfluxDBClient,Pointfrominfluxdb_client.client.write_apiimportSYNCHRONOUSfromdatetimeimportdatetime# 从业务API采集数据 defcollect_business_metrics():采集业务指标# 假设有个内部API返回统计数据responserequests.get(http://internal-api/statistics,timeout10)dataresponse.json()return{page_views:data.get(page_views,0),active_users:data.get(active_users,0),orders:data.get(orders,0),revenue:data.get(revenue,0.0),}# 写入InfluxDB clientInfluxDBClient(urlINFLUX_URL,tokenINFLUX_TOKEN,orgINFLUX_ORG)write_apiclient.write_api(write_optionsSYNCHRONOUS)metricscollect_business_metrics()pointPoint(business)\.tag(platform,main-site)\.field(page_views,metrics[page_views])\.field(active_users,metrics[active_users])\.field(orders,metrics[orders])\.field(revenue,metrics[revenue])\.time(datetime.utcnow())write_api.write(bucketbusiness-metrics,orgINFLUX_ORG,recordpoint)print(f业务指标写入成功PV{metrics[page_views]}, 订单{metrics[orders]})client.close()2.5 查询历史数据做聚合步骤③用Flux语言查询InfluxDB 2.x使用Flux查询语言不是SQL但语法接近frominfluxdb_clientimportInfluxDBClient clientInfluxDBClient(urlINFLUX_URL,tokenINFLUX_TOKEN,orgINFLUX_ORG)query_apiclient.query_api()# 查询最近1小时的平均CPU使用率 query from(bucket: server-metrics) | range(start: -1h) | filter(fn: (r) r[_measurement] cpu) | filter(fn: (r) r[_field] usage_percent) | filter(fn: (r) r[host] web-server-01) | aggregateWindow(every: 5m, fn: mean, createEmpty: false) | yield(name: mean) tablesquery_api.query(query,orgINFLUX_ORG)# 解析结果cpu_data[]fortableintables:forrecordintable.records:cpu_data.append({time:record.get_time().strftime(%H:%M),value:round(record.get_value(),1),})print(f最近1小时CPU数据{cpu_data})# 查询今日峰值 query_max from(bucket: server-metrics) | range(start: today()) | filter(fn: (r) r[_measurement] cpu) | filter(fn: (r) r[_field] usage_percent) | max() resultquery_api.query(query_max,orgINFLUX_ORG)fortableinresult:forrecordintable.records:print(f今日CPU峰值{record.get_value()}% at{record.get_time()})client.close()2.6 超标告警步骤④⑤阈值判断 推送在影刀中拿到CPU数据后【条件判断】 条件cpu_data中最新值 80 → 是 【发送企业微信消息】 内容⚠️ 服务器 web-server-01 CPU使用率超过80% 当前值{latest_value}% 时间{current_time} 请及时检查 → 否结束流程2.7 配合Grafana出图InfluxDB存储数据后在Grafana中添加InfluxDB数据源创建Dashboard配置查询语句设置自动刷新如10秒影刀负责写入数据Grafana负责可视化展示各司其职。三、有什么坑坑1tag和field用反了InfluxDB中tag和field有本质区别特性TagField是否索引是查询快否数据类型只能是字符串数值/字符串/布尔能否聚合不能能基数要求低基数不要放用户ID无限制错误示范把用户ID放到tag中 → 几百万用户就要几百万个tag组合 → 内存爆炸。TEMU店群矩阵自动化运营核价报活动# ❌ 错误高基数字段做tagPoint(order).tag(order_id,ORD-20240001)# 每个订单唯一# ✅ 正确低基数做tag高基数做fieldPoint(order).tag(status,paid).field(order_id,ORD-20240001)坑2时间戳精度不一致InfluxDB接受纳秒精度的时间戳。如果传的是秒级查询时会发现数据都挤在同一秒。fromdatetimeimportdatetime,timezone# ✅ 正确直接用datetime对象库会自动转为纳秒.time(datetime.utcnow())# ❌ 错误手动传Unix秒.time(1234567890)# 这会被当成纳秒级 → 1970年坑3Flux查询语法门槛高Flux不是SQL语法差异大。新手经常写出低效查询。常用Flux模板// 按时间聚合每5分钟平均值 from(bucket: server-metrics) | range(start: -1h) | filter(fn: (r) r._measurement cpu) | aggregateWindow(every: 5m, fn: mean) // 计算环比增长率 from(bucket: business-metrics) | range(start: -2h) | filter(fn: (r) r._measurement business) | derivative(unit: 1h, nonNegative: false) // Top N分析占用最高的5台机器 from(bucket: server-metrics) | range(start: -5m) | filter(fn: (r) r._measurement cpu) | group(columns: [host]) | mean() | sort(columns: [_value], desc: true) | limit(n: 5)坑4数据保留策略InfluxDB默认永久保留数据。如果不设保留策略磁盘会满。# 在InfluxDB中设置bucket的保留策略# 保留30天的数据超过自动删除influx bucket update--nameserver-metrics--retention30d坑5写入性能瓶颈如果影刀每5分钟采集100台服务器的指标每秒写入量不大。但如果采集频率高如每秒写入应该用批量写入# 聚合多个数据点后一次性写入points[]forserverinservers:points.append(Point(cpu).tag(host,server).field(usage,collect(server)))write_api.write(bucketINFLUX_BUCKET,orgINFLUX_ORG,recordpoints)总结InfluxDB 影刀 是运维监控的黄金搭档。影刀负责「定时采集 超标告警」InfluxDB负责「高效存储 时序查询」Grafana负责「可视化看板」。tag和field别用反时间戳别手动传数字数据保留策略要设。