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_MODE=setup\-eDOCKER_INFLUXDB_INIT_USERNAME=admin\-eDOCKER_INFLUXDB_INIT_PASSWORD=admin123456\-eDOCKER_INFLUXDB_INIT_ORG=iot-platform\-eDOCKER_INFLUXDB_INIT_BUCKET=cabinet_data\influxdb:2.7

参数说明：

  • -p 8086:8086：映射 HTTP API 端口，后续所有读写操作都通过这个端口
  • -v influxdb-data:/var/lib/influxdb2：数据持久化，容器删了数据还在
  • DOCKER_INFLUXDB_INIT_MODE=setup：首次启动自动完成初始化
  • DOCKER_INFLUXDB_INIT_ORG=iot-platform：组织名称（相当于一个租户）
  • DOCKER_INFLUXDB_INIT_BUCKET=cabinet_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

创建 Bucket

Bucket 是 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 Token

Token 是操作 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-XPOST"http://localhost:8086/api/v2/write?org=iot-platform&bucket=cabinet_data&precision=s"\-H"Authorization: Token YOUR_API_TOKEN"\-H"Content-Type: text/plain"\-d'device_metrics,device_id=machine-001,workshop=A temp=45.2,current=12.5'# 批量写入（用换行符分隔多条记录）curl-XPOST"http://localhost:8086/api/v2/write?org=iot-platform&bucket=cabinet_data&precision=s"\-H"Authorization: Token YOUR_API_TOKEN"\-H"Content-Type: text/plain"\-d'device_metrics,device_id=machine-001,workshop=A temp=45.2,current=12.5 device_metrics,device_id=machine-001,workshop=A temp=45.5,current=12.7 device_metrics,device_id=machine-002,workshop=A temp=38.1,current=9.8 device_metrics,device_id=machine-003,workshop=B temp=52.3,current=15.2'

几个关键参数：

  • precision=s：时间戳精度，s表示秒。支持ns（纳秒）、us（微秒）、ms（毫秒）、s（秒）
  • /api/v2/write：2.x 的标准写入端点
  • 数据体用换行\n分隔多条记录
  • 一定要把多条记录拼成一个大请求发送，而不是循环发单条。批量写入的吞吐量可以差10倍以上

方式二：influx CLI 写入

CLI 适合快速测试和脚本编写：

# 进入容器dockerexec-itinfluxdbbash# 写入数据influxwrite\--bucketcabinet_data\--precisions\'device_metrics,device_id=machine-001,workshop=A temp=45.2,current=12.5'

批量导入 CSV 文件也很方便：

# 准备 data.csv（格式：measurement,tag key=val field=val timestamp）cat>data.csv<<'EOF' #datatype measurement,tag,long,double dateTime:RFC3339 device_metrics,device_id=machine-001,workshop=A,current,12.5,temp,45.2,2026-07-30T10:00:00Z device_metrics,device_id=machine-001,workshop=A,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-client
frominfluxdb_clientimportInfluxDBClient,Point,WriteOptionsfrominfluxdb_client.client.write_apiimportSYNCHRONOUSimporttimeimportrandom# 连接配置client=InfluxDBClient(url="http://localhost:8086",token="YOUR_API_TOKEN",org="iot-platform")write_api=client.write_api(write_options=SYNCHRONOUS)# 写入单条数据point=Point("device_metrics")\.tag("device_id","machine-001")\.tag("workshop","A")\.field("temp",45.2)\.field("current",12.5)\.time(int(time.time()),write_precision="s")write_api.write(bucket="cabinet_data",record=point)print("单条写入完成")

批量写入（生产环境推荐方式）：

frominfluxdb_clientimportInfluxDBClient,Pointfrominfluxdb_client.client.write_apiimportASYNCHRONOUSimporttimeimportrandom client=InfluxDBClient(url="http://localhost:8086",token="YOUR_API_TOKEN",org="iot-platform")# ASYNCHRONOUS 模式：批量攒够一定量再发送，吞吐量最高write_api=client.write_api(write_options=ASYNCHRONOUS)# 模拟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_time=int(time.time())-1800# 从30分钟前开始success_count=0foroffsetinrange(0,1800,10):# 30分钟，每10秒timestamp=start_time+offsetfordeviceindevices:# 加上随机波动，模拟真实传感器temp=device["base_temp"]+random.uniform(-1.5,1.5)current=device["base_current"]+random.uniform(-0.5,0.5)point=Point("device_metrics")\.tag("device_id",device["id"])\.tag("workshop",device["workshop"])\.field("temp",round(temp,1))\.field("current",round(current,1))\.time(timestamp,write_precision="s")write_api.write(bucket="cabinet_data",record=point)success_count+=1# 确保异步写入完成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_id='machine-001'ANDtime>now()-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 中，写入的多个 field（temp、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 查出来了。下一篇我们深入聚合统计——怎么按小时、按天汇总数据，怎么做下采样和连续查询，让历史数据真正发挥价值。