InfluxDB 安装与实操库、表、数据写入与查询大家好我是黒漂技术佬。前两篇讲了为什么和是什么这篇终于到了动手环节。我们从零开始把 InfluxDB 跑起来然后往里灌数据、查数据全程实操跟紧步伐。InfluxDB 安装Docker 一把梭安装 InfluxDB 最简单的方式就是 Docker。一条命令搞定没有依赖地狱不需要配置环境变量更不用担心把系统搞脏了。Docker 安装# 拉取官方镜像并启动2.7 是当前稳定的 2.x 版本dockerrun-d\--nameinfluxdb\-p8086:8086\-vinfluxdb-data:/var/lib/influxdb2\-vinfluxdb-config:/etc/influxdb2\-eDOCKER_INFLUXDB_INIT_MODEsetup\-eDOCKER_INFLUXDB_INIT_USERNAMEadmin\-eDOCKER_INFLUXDB_INIT_PASSWORDadmin123456\-eDOCKER_INFLUXDB_INIT_ORGiot-platform\-eDOCKER_INFLUXDB_INIT_BUCKETcabinet_data\influxdb:2.7参数说明-p 8086:8086映射 HTTP API 端口后续所有读写操作都通过这个端口-v influxdb-data:/var/lib/influxdb2数据持久化容器删了数据还在DOCKER_INFLUXDB_INIT_MODEsetup首次启动自动完成初始化DOCKER_INFLUXDB_INIT_ORGiot-platform组织名称相当于一个租户DOCKER_INFLUXDB_INIT_BUCKETcabinet_data自动创建名为 cabinet_data 的 bucket启动后访问http://localhost:8086用admin / admin123456登录就能看到 InfluxDB 2.x 自带的管理界面了。这个 UI 可以直接写 Flux 查询、创建 Bucket、管理 Token功能相当齐全。二进制安装备选如果你不喜欢 Docker也可以在官方下载页获取对应系统的二进制包。Windows 用户直接解压把influxd.exe放在 PATH 下即可。启动命令influxd --reporting-disabled--reporting-disabled关闭遥测数据上报。第一次启动后打开http://localhost:8086按向导完成初始化即可。环境准备创建 Bucket 和 Token创建 BucketBucket 是 InfluxDB 2.x 中数据存储的物理容器自带保留策略。在 UI 界面创建很简单Data → Buckets → Create Bucket填入名称和保留时长。也可以用 CLI 创建# 进入容器dockerexec-itinfluxdbbash# 创建 bucket保留30天influx bucket create\--namecabinet_data\--retention30d\--orgiot-platform实战建议生产环境通常会创建两个 Bucket——cabinet_raw原始数据保留7天和cabinet_downsampled下采样数据保留365天。原始数据用于短期排查下采样数据用于长期趋势分析。生成 API TokenToken 是操作 InfluxDB 的凭证。所有 API 请求都需要在 Header 中带上 Token。在 UI 的Data → Tokens → Generate Token → Read/Write Token生成一个记下来后面会用到。# 或者用 CLIinflux auth create\--orgiot-platform\--read-bucket 00xx...\--write-bucket 00xx...数据写入三种方式环境就绪开始往里灌数据。我们模拟一个工控场景——某工厂有3台设备每台每10秒上报温度和电流数据。方式一HTTP API 写入最通用、最灵活的方式任何语言、任何平台都能调用。# 单条写入curl-XPOSThttp://localhost:8086/api/v2/write?orgiot-platformbucketcabinet_dataprecisions\-HAuthorization: Token YOUR_API_TOKEN\-HContent-Type: text/plain\-ddevice_metrics,device_idmachine-001,workshopA temp45.2,current12.5# 批量写入用换行符分隔多条记录curl-XPOSThttp://localhost:8086/api/v2/write?orgiot-platformbucketcabinet_dataprecisions\-HAuthorization: Token YOUR_API_TOKEN\-HContent-Type: text/plain\-ddevice_metrics,device_idmachine-001,workshopA temp45.2,current12.5 device_metrics,device_idmachine-001,workshopA temp45.5,current12.7 device_metrics,device_idmachine-002,workshopA temp38.1,current9.8 device_metrics,device_idmachine-003,workshopB temp52.3,current15.2几个关键参数precisions时间戳精度s表示秒。支持ns纳秒、us微秒、ms毫秒、s秒/api/v2/write2.x 的标准写入端点数据体用换行\n分隔多条记录一定要把多条记录拼成一个大请求发送而不是循环发单条。批量写入的吞吐量可以差10倍以上方式二influx CLI 写入CLI 适合快速测试和脚本编写# 进入容器dockerexec-itinfluxdbbash# 写入数据influxwrite\--bucketcabinet_data\--precisions\device_metrics,device_idmachine-001,workshopA temp45.2,current12.5批量导入 CSV 文件也很方便# 准备 data.csv格式measurement,tag keyval fieldval timestampcatdata.csvEOF #datatype measurement,tag,long,double dateTime:RFC3339 device_metrics,device_idmachine-001,workshopA,current,12.5,temp,45.2,2026-07-30T10:00:00Z device_metrics,device_idmachine-001,workshopA,current,12.7,temp,45.5,2026-07-30T10:00:10Z EOF# 导入influxwrite--bucketcabinet_data--filedata.csv方式三Python influxdb-client 写入实战首选工控、IoT 场景下Python 是最常用的语言。用influxdb-client库写入pipinstallinfluxdb-clientfrominfluxdb_clientimportInfluxDBClient,Point,WriteOptionsfrominfluxdb_client.client.write_apiimportSYNCHRONOUSimporttimeimportrandom# 连接配置clientInfluxDBClient(urlhttp://localhost:8086,tokenYOUR_API_TOKEN,orgiot-platform)write_apiclient.write_api(write_optionsSYNCHRONOUS)# 写入单条数据pointPoint(device_metrics)\.tag(device_id,machine-001)\.tag(workshop,A)\.field(temp,45.2)\.field(current,12.5)\.time(int(time.time()),write_precisions)write_api.write(bucketcabinet_data,recordpoint)print(单条写入完成)批量写入生产环境推荐方式frominfluxdb_clientimportInfluxDBClient,Pointfrominfluxdb_client.client.write_apiimportASYNCHRONOUSimporttimeimportrandom clientInfluxDBClient(urlhttp://localhost:8086,tokenYOUR_API_TOKEN,orgiot-platform)# ASYNCHRONOUS 模式批量攒够一定量再发送吞吐量最高write_apiclient.write_api(write_optionsASYNCHRONOUS)# 模拟3台设备、每10秒上报一次、持续30分钟devices[{id:machine-001,workshop:A,base_temp:45,base_current:12},{id:machine-002,workshop:A,base_temp:38,base_current:9},{id:machine-003,workshop:B,base_temp:52,base_current:15},]start_timeint(time.time())-1800# 从30分钟前开始success_count0foroffsetinrange(0,1800,10):# 30分钟每10秒timestampstart_timeoffsetfordeviceindevices:# 加上随机波动模拟真实传感器tempdevice[base_temp]random.uniform(-1.5,1.5)currentdevice[base_current]random.uniform(-0.5,0.5)pointPoint(device_metrics)\.tag(device_id,device[id])\.tag(workshop,device[workshop])\.field(temp,round(temp,1))\.field(current,round(current,1))\.time(timestamp,write_precisions)write_api.write(bucketcabinet_data,recordpoint)success_count1# 确保异步写入完成write_api.close()print(f写入完成共写入{success_count}条数据)# 预期输出写入完成共写入 540 条数据# (30分钟 × 60秒 ÷ 10秒间隔 × 3台设备 540条)WriteOptions 三种模式SYNCHRONOUS逐条发送响应最慢适合少量数据ASYNCHRONOUS批量缓冲后发送默认每5000条或每1秒发一次推荐BATCHING同 ASYNCHRONOUS可以精确控制批次大小和间隔数据查询Flux 查询语言入门数据写进去了总得查出来看看。InfluxDB 2.x 使用Flux作为查询语言。Flux 是函数式、管道式的语言用|把数据像流水线一样从一个函数传到下一个。基础查询结构from(bucket: cabinet_data) // 1. 指定数据源 | range(start: -30m) // 2. 指定时间范围 | filter(fn: (r) r[_measurement] device_metrics) // 3. 过滤 measurement | filter(fn: (r) r[device_id] machine-001) // 4. 过滤 tag | filter(fn: (r) r[_field] temp) // 5. 过滤 field这个查询相当于 SQL 的SELECTtempFROMdevice_metricsWHEREdevice_idmachine-001ANDtimenow()-30m常用函数查询所有设备最近5分钟的温度from(bucket: cabinet_data) | range(start: -5m) | filter(fn: (r) r[_measurement] device_metrics) | filter(fn: (r) r[_field] temp)查询 machine-001 过去1小时的平均温度聚合from(bucket: cabinet_data) | range(start: -1h) | filter(fn: (r) r[_measurement] device_metrics) | filter(fn: (r) r[device_id] machine-001) | filter(fn: (r) r[_field] temp) | mean()按分钟聚合所有设备的平均电流from(bucket: cabinet_data) | range(start: -30m) | filter(fn: (r) r[_measurement] device_metrics) | filter(fn: (r) r[_field] current) | aggregateWindow(every: 1m, fn: mean)查 machine-001 过去1小时温度的最大值from(bucket: cabinet_data) | range(start: -1h) | filter(fn: (r) r[_measurement] device_metrics) | filter(fn: (r) r[device_id] machine-001) | filter(fn: (r) r[_field] temp) | max()理解_field的用法在 Flux 中写入的多个 fieldtemp、current在查询时会被展开成独立的行每行有一个_field列表示字段名、一个_value列表示字段值。所以当你filter(fn: (r) r[_field] temp)时实际上是从展开的结果中只保留 temp 对应的行。数据库管理查看 Bucket 列表dockerexecinfluxdb influx bucket list--orgiot-platform修改保留策略influx bucket update\--namecabinet_data\--retention7d# 改为保留7天删除 Bucket慎用influx bucket delete--namecabinet_data--orgiot-platform时序数据写入性能实测我写了一个小脚本测试本地 Docker 的写入性能。单线程批量写入device_metrics每次1000条重复1000次共100万条。结果如下写入总量1,000,000 条 耗时3.2秒 吞吐量约 312,500 条/秒31万点/秒这个数字对单节点 InfluxDB 来说很轻松。实际生产环境中如果启用多线程写入、使用 SSD、合理设置 batch 大小单节点轻松跑到 50万~100万点/秒。对比关系型数据库同场景下的几万条/秒差了一个数量级。性能损失场景如果你发现写入速度骤降大概率是 tag 基数爆炸了。检查是否误将高基数字段如订单号、用户ID设成了 tag。到这个节点你应该已经能把 InfluxDB 跑起来、往里写数据、用 Flux 查出来了。下一篇我们深入聚合统计——怎么按小时、按天汇总数据怎么做下采样和连续查询让历史数据真正发挥价值。 
长河君 · 2026/8/15 13:33:38 · 阅读 1.2k 03-InfluxDB 安装与实操：库、表、数据写入与查询
InfluxDB 安装与实操库、表、数据写入与查询大家好我是黒漂技术佬。前两篇讲了为什么和是什么这篇终于到了动手环节。我们从零开始把 InfluxDB 跑起来然后往里灌数据、查数据全程实操跟紧步伐。InfluxDB 安装Docker 一把梭安装 InfluxDB 最简单的方式就是 Docker。一条命令搞定没有依赖地狱不需要配置环境变量更不用担心把系统搞脏了。Docker 安装# 拉取官方镜像并启动2.7 是当前稳定的 2.x 版本dockerrun-d\--nameinfluxdb\-p8086:8086\-vinfluxdb-data:/var/lib/influxdb2\-vinfluxdb-config:/etc/influxdb2\-eDOCKER_INFLUXDB_INIT_MODEsetup\-eDOCKER_INFLUXDB_INIT_USERNAMEadmin\-eDOCKER_INFLUXDB_INIT_PASSWORDadmin123456\-eDOCKER_INFLUXDB_INIT_ORGiot-platform\-eDOCKER_INFLUXDB_INIT_BUCKETcabinet_data\influxdb:2.7参数说明-p 8086:8086映射 HTTP API 端口后续所有读写操作都通过这个端口-v influxdb-data:/var/lib/influxdb2数据持久化容器删了数据还在DOCKER_INFLUXDB_INIT_MODEsetup首次启动自动完成初始化DOCKER_INFLUXDB_INIT_ORGiot-platform组织名称相当于一个租户DOCKER_INFLUXDB_INIT_BUCKETcabinet_data自动创建名为 cabinet_data 的 bucket启动后访问http://localhost:8086用admin / admin123456登录就能看到 InfluxDB 2.x 自带的管理界面了。这个 UI 可以直接写 Flux 查询、创建 Bucket、管理 Token功能相当齐全。二进制安装备选如果你不喜欢 Docker也可以在官方下载页获取对应系统的二进制包。Windows 用户直接解压把influxd.exe放在 PATH 下即可。启动命令influxd --reporting-disabled--reporting-disabled关闭遥测数据上报。第一次启动后打开http://localhost:8086按向导完成初始化即可。环境准备创建 Bucket 和 Token创建 BucketBucket 是 InfluxDB 2.x 中数据存储的物理容器自带保留策略。在 UI 界面创建很简单Data → Buckets → Create Bucket填入名称和保留时长。也可以用 CLI 创建# 进入容器dockerexec-itinfluxdbbash# 创建 bucket保留30天influx bucket create\--namecabinet_data\--retention30d\--orgiot-platform实战建议生产环境通常会创建两个 Bucket——cabinet_raw原始数据保留7天和cabinet_downsampled下采样数据保留365天。原始数据用于短期排查下采样数据用于长期趋势分析。生成 API TokenToken 是操作 InfluxDB 的凭证。所有 API 请求都需要在 Header 中带上 Token。在 UI 的Data → Tokens → Generate Token → Read/Write Token生成一个记下来后面会用到。# 或者用 CLIinflux auth create\--orgiot-platform\--read-bucket 00xx...\--write-bucket 00xx...数据写入三种方式环境就绪开始往里灌数据。我们模拟一个工控场景——某工厂有3台设备每台每10秒上报温度和电流数据。方式一HTTP API 写入最通用、最灵活的方式任何语言、任何平台都能调用。# 单条写入curl-XPOSThttp://localhost:8086/api/v2/write?orgiot-platformbucketcabinet_dataprecisions\-HAuthorization: Token YOUR_API_TOKEN\-HContent-Type: text/plain\-ddevice_metrics,device_idmachine-001,workshopA temp45.2,current12.5# 批量写入用换行符分隔多条记录curl-XPOSThttp://localhost:8086/api/v2/write?orgiot-platformbucketcabinet_dataprecisions\-HAuthorization: Token YOUR_API_TOKEN\-HContent-Type: text/plain\-ddevice_metrics,device_idmachine-001,workshopA temp45.2,current12.5 device_metrics,device_idmachine-001,workshopA temp45.5,current12.7 device_metrics,device_idmachine-002,workshopA temp38.1,current9.8 device_metrics,device_idmachine-003,workshopB temp52.3,current15.2几个关键参数precisions时间戳精度s表示秒。支持ns纳秒、us微秒、ms毫秒、s秒/api/v2/write2.x 的标准写入端点数据体用换行\n分隔多条记录一定要把多条记录拼成一个大请求发送而不是循环发单条。批量写入的吞吐量可以差10倍以上方式二influx CLI 写入CLI 适合快速测试和脚本编写# 进入容器dockerexec-itinfluxdbbash# 写入数据influxwrite\--bucketcabinet_data\--precisions\device_metrics,device_idmachine-001,workshopA temp45.2,current12.5批量导入 CSV 文件也很方便# 准备 data.csv格式measurement,tag keyval fieldval timestampcatdata.csvEOF #datatype measurement,tag,long,double dateTime:RFC3339 device_metrics,device_idmachine-001,workshopA,current,12.5,temp,45.2,2026-07-30T10:00:00Z device_metrics,device_idmachine-001,workshopA,current,12.7,temp,45.5,2026-07-30T10:00:10Z EOF# 导入influxwrite--bucketcabinet_data--filedata.csv方式三Python influxdb-client 写入实战首选工控、IoT 场景下Python 是最常用的语言。用influxdb-client库写入pipinstallinfluxdb-clientfrominfluxdb_clientimportInfluxDBClient,Point,WriteOptionsfrominfluxdb_client.client.write_apiimportSYNCHRONOUSimporttimeimportrandom# 连接配置clientInfluxDBClient(urlhttp://localhost:8086,tokenYOUR_API_TOKEN,orgiot-platform)write_apiclient.write_api(write_optionsSYNCHRONOUS)# 写入单条数据pointPoint(device_metrics)\.tag(device_id,machine-001)\.tag(workshop,A)\.field(temp,45.2)\.field(current,12.5)\.time(int(time.time()),write_precisions)write_api.write(bucketcabinet_data,recordpoint)print(单条写入完成)批量写入生产环境推荐方式frominfluxdb_clientimportInfluxDBClient,Pointfrominfluxdb_client.client.write_apiimportASYNCHRONOUSimporttimeimportrandom clientInfluxDBClient(urlhttp://localhost:8086,tokenYOUR_API_TOKEN,orgiot-platform)# ASYNCHRONOUS 模式批量攒够一定量再发送吞吐量最高write_apiclient.write_api(write_optionsASYNCHRONOUS)# 模拟3台设备、每10秒上报一次、持续30分钟devices[{id:machine-001,workshop:A,base_temp:45,base_current:12},{id:machine-002,workshop:A,base_temp:38,base_current:9},{id:machine-003,workshop:B,base_temp:52,base_current:15},]start_timeint(time.time())-1800# 从30分钟前开始success_count0foroffsetinrange(0,1800,10):# 30分钟每10秒timestampstart_timeoffsetfordeviceindevices:# 加上随机波动模拟真实传感器tempdevice[base_temp]random.uniform(-1.5,1.5)currentdevice[base_current]random.uniform(-0.5,0.5)pointPoint(device_metrics)\.tag(device_id,device[id])\.tag(workshop,device[workshop])\.field(temp,round(temp,1))\.field(current,round(current,1))\.time(timestamp,write_precisions)write_api.write(bucketcabinet_data,recordpoint)success_count1# 确保异步写入完成write_api.close()print(f写入完成共写入{success_count}条数据)# 预期输出写入完成共写入 540 条数据# (30分钟 × 60秒 ÷ 10秒间隔 × 3台设备 540条)WriteOptions 三种模式SYNCHRONOUS逐条发送响应最慢适合少量数据ASYNCHRONOUS批量缓冲后发送默认每5000条或每1秒发一次推荐BATCHING同 ASYNCHRONOUS可以精确控制批次大小和间隔数据查询Flux 查询语言入门数据写进去了总得查出来看看。InfluxDB 2.x 使用Flux作为查询语言。Flux 是函数式、管道式的语言用|把数据像流水线一样从一个函数传到下一个。基础查询结构from(bucket: cabinet_data) // 1. 指定数据源 | range(start: -30m) // 2. 指定时间范围 | filter(fn: (r) r[_measurement] device_metrics) // 3. 过滤 measurement | filter(fn: (r) r[device_id] machine-001) // 4. 过滤 tag | filter(fn: (r) r[_field] temp) // 5. 过滤 field这个查询相当于 SQL 的SELECTtempFROMdevice_metricsWHEREdevice_idmachine-001ANDtimenow()-30m常用函数查询所有设备最近5分钟的温度from(bucket: cabinet_data) | range(start: -5m) | filter(fn: (r) r[_measurement] device_metrics) | filter(fn: (r) r[_field] temp)查询 machine-001 过去1小时的平均温度聚合from(bucket: cabinet_data) | range(start: -1h) | filter(fn: (r) r[_measurement] device_metrics) | filter(fn: (r) r[device_id] machine-001) | filter(fn: (r) r[_field] temp) | mean()按分钟聚合所有设备的平均电流from(bucket: cabinet_data) | range(start: -30m) | filter(fn: (r) r[_measurement] device_metrics) | filter(fn: (r) r[_field] current) | aggregateWindow(every: 1m, fn: mean)查 machine-001 过去1小时温度的最大值from(bucket: cabinet_data) | range(start: -1h) | filter(fn: (r) r[_measurement] device_metrics) | filter(fn: (r) r[device_id] machine-001) | filter(fn: (r) r[_field] temp) | max()理解_field的用法在 Flux 中写入的多个 fieldtemp、current在查询时会被展开成独立的行每行有一个_field列表示字段名、一个_value列表示字段值。所以当你filter(fn: (r) r[_field] temp)时实际上是从展开的结果中只保留 temp 对应的行。数据库管理查看 Bucket 列表dockerexecinfluxdb influx bucket list--orgiot-platform修改保留策略influx bucket update\--namecabinet_data\--retention7d# 改为保留7天删除 Bucket慎用influx bucket delete--namecabinet_data--orgiot-platform时序数据写入性能实测我写了一个小脚本测试本地 Docker 的写入性能。单线程批量写入device_metrics每次1000条重复1000次共100万条。结果如下写入总量1,000,000 条 耗时3.2秒 吞吐量约 312,500 条/秒31万点/秒这个数字对单节点 InfluxDB 来说很轻松。实际生产环境中如果启用多线程写入、使用 SSD、合理设置 batch 大小单节点轻松跑到 50万~100万点/秒。对比关系型数据库同场景下的几万条/秒差了一个数量级。性能损失场景如果你发现写入速度骤降大概率是 tag 基数爆炸了。检查是否误将高基数字段如订单号、用户ID设成了 tag。到这个节点你应该已经能把 InfluxDB 跑起来、往里写数据、用 Flux 查出来了。下一篇我们深入聚合统计——怎么按小时、按天汇总数据怎么做下采样和连续查询让历史数据真正发挥价值。