InfluxQL + Flux 双语言实战：数据写入、查询与聚合一篇搞定

# 写入 3 台服务器的 CPU 和内存数据（Line Protocol 格式）
server_metric,host=web-01,cpu=0 cpu_usage=45.2,mem_usage=62.1 1689123456000000000
server_metric,host=web-01,cpu=1 cpu_usage=32.8,mem_usage=62.1 1689123456000000000
server_metric,host=web-02,cpu=0 cpu_usage=78.5,mem_usage=85.3 1689123456000000000
server_metric,host=web-02,cpu=1 cpu_usage=65.1,mem_usage=85.3 1689123456000000000
server_metric,host=db-01,cpu=0 cpu_usage=23.4,mem_usage=45.6 1689123456000000000

curl -X POST "http://localhost:8086/api/v2/write?org=myorg&bucket=mydb&precision=ns" \
  -H "Authorization: Token MY_TOKEN" \
  -H "Content-Type: text/plain; charset=utf-8" \
  -d 'server_metric,host=web-01,cpu=0 cpu_usage=45.2,mem_usage=62.1 1689123456000000000'

# 批量写入：每条数据一行，用换行符分隔
curl -X POST "http://localhost:8086/api/v2/write?org=myorg&bucket=mydb&precision=ns" \
  -H "Authorization: Token MY_TOKEN" \
  -H "Content-Type: text/plain; charset=utf-8" \
  -d 'server_metric,host=web-01,cpu=0 cpu_usage=45.2,mem_usage=62.1 1689123456000000000
server_metric,host=web-01,cpu=1 cpu_usage=32.8,mem_usage=62.1 1689123456000000010
server_metric,host=web-02,cpu=0 cpu_usage=78.5,mem_usage=85.3 1689123456000000020'

# Python 批量写入封装
import requests
import time
def batch_write(metrics, batch_size=500):
    """批量写入 InfluxDB，每批 500 条，批次间间隔 10ms 控制速率"""
    url = "http://localhost:8086/api/v2/write?org=myorg&bucket=mydb&precision=ns"
    headers = {
        "Authorization": "Token MY_TOKEN",
        "Content-Type": "text/plain; charset=utf-8"
    }
    for i in range(0, len(metrics), batch_size):
        batch = metrics[i:i + batch_size]
        data = "\n".join(batch)
        resp = requests.post(url, headers=headers, data=data)
        if resp.status_code != 204:
            print(f"[ERROR] Batch {i//batch_size} failed: {resp.status_code}")
        time.sleep(0.01)  # 控制写入速率，避免打满带宽

# 统一使用秒级时间戳
curl -X POST "...&precision=s" -d 'server_metric cpu=45.2 1689123456'

# Telegraf 配置中指定精度
# 不要指定 precision，让 Telegraf 自动使用纳秒
[[outputs.influxdb_v2]]
  urls = ["http://localhost:8086"]
  token = "MY_TOKEN"
  organization = "myorg"
  bucket = "mydb"
  content_encoding = "gzip"

-- InfluxQL 写法：类 SQL 语法，上手成本低
SELECT MEAN(cpu_usage)
FROM server_metric
WHERE host = 'web-01'
  AND time > now() - 1h
GROUP BY cpu;

// Flux 写法：函数式管道语法，表达能力更强
from(bucket: "mydb")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "server_metric" and r.host == "web-01")
  |> filter(fn: (r) => r._field == "cpu_usage")
  |> group(columns: ["cpu"])
  |> mean()
  |> yield(name: "mean_cpu")

-- InfluxQL：GROUP BY time 实现时间桶聚合
SELECT MEAN(cpu_usage)
FROM server_metric
WHERE time > now() - 24h
  AND host =~ /web-/
GROUP BY time(1h), host
  FILL(null);

// Flux：aggregateWindow 实现时间窗口聚合
from(bucket: "mydb")
  |> range(start: -24h)
  |> filter(fn: (r) => r._measurement == "server_metric")
  |> filter(fn: (r) => r.host =~ /web-/)
  |> filter(fn: (r) => r._field == "cpu_usage")
  |> aggregateWindow(every: 1h, fn: mean, createEmpty: false)
  |> yield(name: "hourly_cpu")

// Flux 的优势：多 field 联合查询 + JOIN
cpu = from(bucket: "mydb")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "server_metric" and r.host == "web-01")
  |> filter(fn: (r) => r._field == "cpu_usage")
  |> aggregateWindow(every: 5m, fn: mean)
mem = from(bucket: "mydb")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "server_metric" and r.host == "web-01")
  |> filter(fn: (r) => r._field == "mem_usage")
  |> aggregateWindow(every: 5m, fn: mean)
// 两个表 JOIN，计算 CPU/内存比值
join(tables: {cpu: cpu, mem: mem}, on: ["_time", "host"])
  |> map(fn: (r) => ({r with ratio: r._value_cpu / r._value_mem}))
  |> yield(name: "cpu_mem_ratio")

// Flux 跨 Measurement 查询：指标 + 告警时间线
metrics = from(bucket: "mydb")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "server_metric")
  |> filter(fn: (r) => r._field == "cpu_usage")
alerts = from(bucket: "mydb")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "alert_event")
  |> filter(fn: (r) => r.severity == "critical")
// 按时间对齐，找出指标异常与告警的关联
join(tables: {metric: metrics, alert: alerts}, on: ["_time", "host"])
  |> map(fn: (r) => ({
    r with
    alert_context: "CPU={r.alert_name}"
  }))
  |> yield(name: "metric_alert_correlation")

# InfluxDB 2.x 中 RP 由 Bucket 的 retention 参数控制
# 创建 Bucket 时指定数据保留 30 天
influx bucket create \
  --name mydb \
  --retention 30d \
  --org myorg

# 查看现有 Bucket 及保留期
influx bucket list
# 修改保留期（已有数据不会被删除，但过期后自动清理）
influx bucket update \
  --id BUCKET_ID \
  --retention 90d

# 紧急设置保留期（已存在的数据不会被删除，但后续会自动清理）
influx bucket update --id BUCKET_ID --retention 30d
# 手动清理旧数据（如果磁盘已经撑不住了）
influx delete --bucket mydb --start 1970-01-01T00:00:00Z --stop 2025-06-01T00:00:00Z
# 查看 Bucket 大小
influx bucket list --org myorg

-- InfluxDB 1.x 方式：Continuous Query（2.x 中通过 Task 实现）
CREATE CONTINUOUS QUERY "cq_hourly_cpu" ON "mydb"
BEGIN
  SELECT MEAN(cpu_usage) AS cpu_usage_hourly
  INTO "hourly_metrics"
  FROM "server_metric"
  GROUP BY time(1h), host, cpu
END

// InfluxDB 2.x 使用 Task 实现降采样
// 在 Web UI 的 Tasks 页面中创建
option task = {
  name: "hourly-downsample",
  every: 1h,          // 每小时执行一次
  offset: 5m          // 延迟 5 分钟执行（等待数据完全到达）
}
from(bucket: "mydb")
  |> range(start: -2h)  // 处理过去 2 小时的数据（保证覆盖）
  |> filter(fn: (r) => r._measurement == "server_metric")
  |> filter(fn: (r) => r._field == "cpu_usage")
  |> aggregateWindow(every: 1h, fn: mean)
  |> set(key: "_measurement", value: "hourly_cpu")
  |> to(bucket: "mydb_downsampled", org: "myorg")

option task = {
  name: "hourly-downsample",
  every: 1h,
  offset: 30m    // 改为 30 分钟偏移，给延迟数据足够的时间
}
// 查询范围也扩大到 3 小时
from(bucket: "mydb")
  |> range(start: -3h)
  // ... 后续聚合逻辑不变

-- 查询时显式指定时区偏移（8h = UTC+8）
SELECT MEAN(cpu_usage)
FROM server_metric
WHERE time > now() - 24h
GROUP BY time(1h, 8h)
  FILL(previous);

# 1. 开启 gzip 压缩（减少网络传输量 60-80%）
curl -X POST "..." -H "Content-Encoding: gzip" --data-binary @data.gz
# 2. 增大 batch size（每次至少 500 条）
# 3. 降低写入精度（从纳秒降到秒级）

# influxdb.conf — 写入调优
[coordinator]
  write-timeout = "10s"
  max-concurrent-queries = 0
[data]
  cache-max-memory-size = "1g"
  cache-snapshot-write-cold-duration = "10m"
  compact-full-write-cold-duration = "30m"

【InfluxQL 适用场景】
□ 简单聚合查询（MEAN/MAX/MIN/COUNT）
□ GROUP BY time 时间桶聚合
□ 简单 WHERE 过滤（Tag 条件 + 时间范围）
□ FILL 填充空值
【Flux 适用场景】
□ 跨 Measurement 查询
□ 多表 JOIN 关联分析
□ 条件分支和自定义函数
□ 数据预处理（过滤/转换/聚合/输出）
【生产必配】
□ Retention Policy：必须设置，默认 0=永久保留是大坑
□ Continuous Query / Task：高频→低频降采样
□ 写入精度统一：所有采集端用同一个 precision