数据写入方式

InfluxDB 的“写入格式”和“写入工具”是两个层次。Line Protocol 定义一条 Point 怎样表示；CLI、HTTP API、Client Library 和 Telegraf 负责采集、组装或传输这些 Point。

---

方式对比

方式	适合场景	优点	注意事项
Line Protocol 文件	批量导入、测试、数据交换	格式直观、跨语言	需要自行发送或交给 CLI
influxdb3 CLI	本地验证、运维脚本、小批量导入	使用简单、便于排错	不适合作为高吞吐应用写入层
HTTP Write API	Shell、网关、自研协议适配	依赖少、控制直接	自行处理批量、重试、超时和 Token
Client Library	Java、Go、Python 等业务应用	提供批处理、序列化和连接管理	版本、Endpoint 和精度要匹配
Telegraf	主机、容器、中间件、MQTT、SNMP 等采集	插件丰富、配置式采集	需要规划缓冲、过滤和字段类型

无论从哪种方式写入，最终都需要明确 Database/Bucket、认证 Token、Table/Measurement、Tag、Field 和 Timestamp。

---

Line Protocol

Line Protocol 是 InfluxDB 的文本写入格式：

<table>[,<tag_key>=<tag_value>...] <field_key>=<field_value>[,<field>...] [timestamp]

示例：

sensor_data,site=hefei,device_id=sensor01 temperature=26.3,humidity=61.5,online=true 1787212800

解析结果：

部分	内容
Table	sensor_data
Tag Set	site=hefei,device_id=sensor01
Field Set	temperature=26.3,humidity=61.5,online=true
Timestamp	1787212800，本例为秒精度

Line Protocol 对空格敏感：第一个未转义空格分隔 Tag Set 与 Field Set，第二个未转义空格分隔 Field Set 与 Timestamp。

类型写法

example float_value=1.5,integer_value=2i,unsigned_value=3u,string_value="ok",boolean_value=true

特殊字符转义

Table、Tag 或 Field Key 中的空格和逗号需要转义，Tag 中的等号也需要转义：

environment,room=Living\ Room temperature=26.3

生产 Schema 更推荐使用小写字母、数字和下划线，减少转义需求。

批量写入

一行表示一个 Point，多行组成一个批次：

sensor_data,site=hefei,device_id=sensor01 temperature=26.3 1787212800
sensor_data,site=hefei,device_id=sensor02 temperature=27.1 1787212800
sensor_data,site=hefei,device_id=sensor01 temperature=27.0 1787213100

批量写入通常比每个 Point 发一次请求更高效，但批次过大会增加单次失败后的重试成本和内存占用。

---

使用 influxdb3 CLI

直接写入一条 Point：

influxdb3 write \
  --host http://127.0.0.1:8181 \
  --database iot \
  --token "$INFLUXDB3_AUTH_TOKEN" \
  --precision s \
  'sensor_data,site=hefei,device_id=sensor01 temperature=26.3,humidity=61.5 1787212800'

从文件批量写入：

influxdb3 write \
  --host http://127.0.0.1:8181 \
  --database iot \
  --token "$INFLUXDB3_AUTH_TOKEN" \
  --precision s \
  --file sensor-data.lp

CLI 适合验证 Token、Database、网络和 Line Protocol 是否正确。应用长期运行时，通常使用 Client Library 或 Telegraf。

---

使用 HTTP API

InfluxDB 3 原生写入端点使用 /api/v3/write_lp。以下示例提交秒精度 Line Protocol：

curl --request POST \
  "http://127.0.0.1:8181/api/v3/write_lp?db=iot&precision=second" \
  --header "Authorization: Bearer $INFLUXDB3_AUTH_TOKEN" \
  --header "Content-Type: text/plain; charset=utf-8" \
  --data-binary 'sensor_data,site=hefei,device_id=sensor01 temperature=26.3,humidity=61.5 1787212800'

发送文件：

curl --request POST \
  "http://127.0.0.1:8181/api/v3/write_lp?db=iot&precision=second" \
  --header "Authorization: Bearer $INFLUXDB3_AUTH_TOKEN" \
  --header "Content-Type: text/plain; charset=utf-8" \
  --data-binary @sensor-data.lp

HTTP 写入需要处理：

• 连接和请求超时；
• 认证失败与 Token 轮换；
• 429、5xx 等可重试错误；
• 语法和类型冲突等不可重试错误；
• 批次大小、压缩和重试退避；
• 重试导致的重复 Point 语义。

InfluxDB 3 还提供兼容 v1/v2 的写入端点，但参数名和精度取值不同。新程序优先使用目标产品推荐的原生 API，迁移旧客户端时再使用兼容端点。

---

使用 Client Library

Client Library 适合业务应用。典型流程是：

业务对象
  -> 转换为 Point 或 Line Protocol
  -> 写入内存批次
  -> 达到条数或时间阈值后发送
  -> 失败时按错误类型重试

伪代码示例：

client = connect(host, token, database)

point = Point("sensor_data")
    .tag("site", "hefei")
    .tag("device_id", "sensor01")
    .field("temperature", 26.3)
    .field("humidity", 61.5)
    .timestamp(observed_at)

client.write(point)

不同语言和 InfluxDB 产品代际的 SDK API 会变化，使用前应确认：

• SDK 是否支持 InfluxDB 3；
• 使用 v3 原生端点还是 v1/v2 兼容端点；
• Database 与旧 SDK 中 Bucket 参数如何映射；
• Timestamp 精度和时区；
• 同步写入还是异步批量写入；
• Flush、关闭客户端和进程退出时的数据排空行为。

高频业务写入一般不要在每条 Point 后立即 Flush。更常见的做法是按条数与时间双阈值批量发送，例如“达到 5,000 条或等待 1 秒即发送”，具体数值需要压测。

---

使用 Telegraf

Telegraf 是插件化采集代理，配置通常由四类插件组成：

Input -> Processor -> Aggregator -> Output

插件	作用	示例
Input	从数据源采集指标	CPU、Docker、Prometheus、MQTT、SNMP
Processor	重命名、过滤、转换 Tag/Field	Converter、Regex、Rename
Aggregator	在本地按窗口聚合	Basic Stats、Histogram
Output	将结果发送到目标系统	InfluxDB v2 Output、HTTP Output

采集主机指标

[agent]
  interval = "10s"
  flush_interval = "10s"
  metric_batch_size = 1000
  metric_buffer_limit = 10000

[[inputs.cpu]]
  percpu = true
  totalcpu = true
  collect_cpu_time = false
  report_active = true

[[inputs.mem]]

[[outputs.influxdb_v2]]
  urls = ["http://127.0.0.1:8181"]
  token = "$INFLUXDB3_AUTH_TOKEN"
  organization = ""
  bucket = "host_metrics"

InfluxDB 3 可通过兼容写入 API 接收 Telegraf 的 InfluxDB v2 Output 数据。bucket 映射到目标 Database；具体产品版本是否需要 organization 以及认证方式，应以对应版本文档为准。

从 MQTT 采集 JSON

设备发布消息：

{
  "device_id": "sensor01",
  "site": "hefei",
  "temperature": 26.3,
  "humidity": 61.5,
  "online": true
}

Telegraf 配置：

[[inputs.mqtt_consumer]]
  servers = ["tcp://127.0.0.1:1883"]
  topics = ["factory/+/sensor"]
  data_format = "json_v2"

  [[inputs.mqtt_consumer.json_v2]]
    measurement_name = "sensor_data"

    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "device_id"

    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "site"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "temperature"
      type = "float"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "humidity"
      type = "float"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "online"
      type = "bool"

[[outputs.influxdb_v2]]
  urls = ["http://127.0.0.1:8181"]
  token = "$INFLUXDB3_AUTH_TOKEN"
  organization = ""
  bucket = "iot"

这里必须显式决定 JSON 属性是 Tag 还是 Field。device_id、site 用于标识和分组，适合作为 Tag；温度、湿度和在线状态是观测值，作为 Field。

测试配置

先让 Telegraf 输出一次采集结果，不发送到 InfluxDB：

telegraf --config telegraf.conf --test

观察输出中的 Measurement、Tag、Field 和类型是否符合预期，再启动常驻进程。尤其要检查整数与浮点类型、时间戳、动态 Tag 和敏感信息。

---

如何选择

临时测试或导入文件
  -> influxdb3 CLI

应用程序自产生业务指标
  -> 官方或兼容 Client Library

已有系统只方便发 HTTP
  -> HTTP Write API

采集主机、容器、中间件、MQTT、SNMP
  -> Telegraf

多来源统一接入、清洗和路由
  -> Telegraf 或独立采集管道，再写 InfluxDB

Telegraf 与 Client Library 也可以组合：应用通过 SDK 写业务指标，Telegraf 采集主机和中间件指标，两类数据进入不同 Table，必要时使用不同 Database 和 Retention 配置。

---

可靠写入原则

• 批量发送 Point，避免每条数据建立一次请求。
• 明确 Timestamp 精度，优先写事件时间。
• Tag、Field 和类型在正式接入前固定下来。
• 为客户端设置内存缓冲上限，避免目标不可用时耗尽内存。
• 区分可重试与不可重试错误，类型冲突不能靠无限重试解决。
• 为批次记录成功、失败、丢弃和重试指标。
• Token 通过环境变量或 Secret 注入，不写入配置仓库。
• 压测批次大小、Flush 周期、并发数和服务端容量。
• 了解目标版本的重复 Point 行为，不把重试等同于可靠覆盖。

参考资料

• InfluxDB 3 写入数据
• InfluxDB 3 Line Protocol
• influxdb3 write
• Telegraf Documentation
• Telegraf InfluxDB v2 Output