性能优化 #

性能优化维度：

写入性能
├── 批量写入
├── 并发控制
├── 数据格式
└── 网络优化

查询性能
├── 查询优化
├── 索引使用
├── 时间范围
└── 聚合策略

存储性能
├── 压缩策略
├── 数据保留
├── Schema设计
└── 磁盘IO

资源配置
├── 内存配置
├── 缓存配置
├── 并发配置
└── 超时配置

关键性能指标：

写入指标
├── 写入吞吐量（points/s）
├── 写入延迟（ms）
├── 写入错误率（%）
└── 批量大小

查询指标
├── 查询延迟（ms）
├── 查询QPS
├── 查询错误率（%）
└── 慢查询比例

资源指标
├── CPU使用率（%）
├── 内存使用率（%）
├── 磁盘IO（MB/s）
└── 网络带宽（MB/s）

# 优化前：单条写入
for point in points:
    write_api.write(bucket="my-bucket", record=point)

# 优化后：批量写入
batch_size = 5000
batch = []
for point in points:
    batch.append(point)
    if len(batch) >= batch_size:
        write_api.write(bucket="my-bucket", record=batch)
        batch = []
if batch:
    write_api.write(bucket="my-bucket", record=batch)

批量写入建议：

批量大小
├── 推荐：5000-10000条
├── 过小：效率低
└── 过大：超时风险

写入频率
├── 推荐：1-5秒/批
├── 实时性要求高：更频繁
└── 批量导入：更大批次

from concurrent.futures import ThreadPoolExecutor
from influxdb_client import InfluxDBClient, Point

def write_batch(client, bucket, points):
    write_api = client.write_api()
    write_api.write(bucket=bucket, record=points)
    write_api.close()

client = InfluxDBClient(url="http://localhost:8086", token="YOUR_TOKEN")

# 并发写入
with ThreadPoolExecutor(max_workers=4) as executor:
    for batch in batches:
        executor.submit(write_batch, client, "my-bucket", batch)

并发写入建议：

并发数
├── 推荐：2-4个连接
├── 过多：服务器压力大
└── 根据服务器能力调整

连接复用
├── 使用连接池
├── 避免频繁创建连接
└── 长连接优先

行协议优化：

Tag设计
├── 使用低基数值
├── 避免高基数Tag
├── Tag数量适中
└── 值长度适中

Field设计
├── 合适的数据类型
├── 避免过多Field
└── 使用数值类型

示例：
# 优化前（高基数Tag）
cpu,host=server-uuid-12345,process=process-uuid-67890 value=78.5

# 优化后（低基数Tag）
cpu,host=server01,process_type=web value=78.5

// 优化前：查询过长时间
from(bucket: "my-bucket")
    |> range(start: -30d)
    |> filter(fn: (r) => r._measurement == "cpu")

// 优化后：缩小时间范围
from(bucket: "my-bucket")
    |> range(start: -1h)
    |> filter(fn: (r) => r._measurement == "cpu")

// 优化前：先过滤值
from(bucket: "my-bucket")
    |> range(start: -1h)
    |> filter(fn: (r) => r._value > 50)
    |> filter(fn: (r) => r._measurement == "cpu")
    |> filter(fn: (r) => r.host == "server01")

// 优化后：先过滤measurement和tag
from(bucket: "my-bucket")
    |> range(start: -1h)
    |> filter(fn: (r) => r._measurement == "cpu")
    |> filter(fn: (r) => r.host == "server01")
    |> filter(fn: (r) => r._value > 50)

// 优化前：全量聚合
from(bucket: "my-bucket")
    |> range(start: -30d)
    |> filter(fn: (r) => r._measurement == "cpu")
    |> mean()

// 优化后：窗口聚合
from(bucket: "my-bucket")
    |> range(start: -30d)
    |> filter(fn: (r) => r._measurement == "cpu")
    |> aggregateWindow(every: 1h, fn: mean)

from influxdb_client import InfluxDBClient
import time

class CachedQuery:
    def __init__(self, client, ttl=60):
        self.client = client
        self.ttl = ttl
        self.cache = {}
    
    def query(self, query, bucket):
        cache_key = f"{query}_{bucket}"
        if cache_key in self.cache:
            result, timestamp = self.cache[cache_key]
            if time.time() - timestamp < self.ttl:
                return result
        
        query_api = self.client.query_api()
        result = query_api.query(query)
        self.cache[cache_key] = (result, time.time())
        return result

Schema设计原则：

Tag设计
├── 用于查询过滤和分组
├── 值具有低基数
├── 不经常变化
└── 示例：host、region、env

Field设计
├── 存储实际数据值
├── 可以是任何类型
├── 不用于分组
└── 示例：value、count、status

避免
├── 高基数Tag（如UUID）
├── 过多Tag（<10个）
├── 过长Tag值
└── 时间戳作为Tag

# 创建Bucket时设置保留时间
influx bucket create \
    --name metrics \
    --retention 720h  # 30天

# 更新保留时间
influx bucket update \
    --id BUCKET_ID \
    --retention 168h  # 7天

保留策略建议：

数据类型          保留时间
────────────────────────────
实时监控数据      7-30天
日志数据          7-14天
业务指标          30-90天
历史数据          365天+

// 创建降采样任务
option task = {
    name: "downsample-metrics",
    every: 1h
}

from(bucket: "metrics-raw")
    |> range(start: -1h)
    |> aggregateWindow(every: 5m, fn: mean)
    |> to(bucket: "metrics-5m")

# config.toml

[storage]
  # 缓存最大内存
  cache-max-memory-size = 1073741824  # 1GB
  
  # WAL缓存大小
  wal-max-size = 52428800  # 50MB
  
  # 索引类型
  index-version = "tsi1"

[coordinator]
  # 最大并发查询
  max-concurrent-queries = 0  # 无限制
  
  # 查询超时
  query-timeout = "0"  # 无限制
  
  # 写入超时
  write-timeout = "10s"

[http]
  # 绑定地址
  bind-address = ":8086"
  
  # 最大请求体大小
  max-body-size = 0  # 无限制
  
  # 连接超时
  idle-timeout = "3m0s"
  
  # 读超时
  read-header-timeout = "10s"

[storage]
  # 数据目录
  data-path = "/var/lib/influxdb2/engine"
  
  # WAL目录
  wal-path = "/var/lib/influxdb2/engine/wal"
  
  # 压缩设置
  compact-full-write-cold-duration = "4h"
  compact-throughput = 50331648

存储建议：

磁盘类型
├── SSD：强烈推荐
├── NVMe：性能最佳
└── HDD：不推荐

RAID配置
├── RAID 10：性能+冗余
├── RAID 5：容量+冗余
└── RAID 0：性能（无冗余）

文件系统
├── XFS：推荐
├── ext4：兼容性好
└── ZFS：高级特性

内存建议：

容量规划
├── 最小：8GB
├── 推荐：16-32GB
└── 高负载：64GB+

内存分配
├── 操作系统：25%
├── InfluxDB：50%
└── 其他服务：25%

监控内存使用
├── 关注RSS
├── 关注缓存
└── 设置告警

CPU建议：

核心数
├── 最小：4核
├── 推荐：8-16核
└── 高负载：32核+

频率
├── 高频率优先
├── 单核性能重要
└── 避免低功耗CPU

监控CPU使用
├── 关注系统CPU
├── 关注用户CPU
└── 关注IO等待

// 监控写入性能
from(bucket: "_internal/monitor")
    |> range(start: -1h)
    |> filter(fn: (r) => r._measurement == "write")
    |> derivative(unit: 1s, nonNegative: true)

// 监控查询性能
from(bucket: "_internal/monitor")
    |> range(start: -1h)
    |> filter(fn: (r) => r._measurement == "query")
    |> group(columns: ["_field"])
    |> mean()

# 查看慢查询日志
cat /var/log/influxdb/influxd.log | jq 'select(.duration > 1000000000)'

# 分析查询模式
cat /var/log/influxdb/influxd.log | jq -r '.query' | sort | uniq -c | sort -rn

# 使用influx-stress测试
influx-stress write \
    --host http://localhost:8086 \
    --token YOUR_TOKEN \
    --bucket my-bucket \
    --points 1000000 \
    --batch-size 10000

写入优化清单：

✓ 使用批量写入
✓ 合理设置批量大小
✓ 使用并发写入
✓ 优化数据格式
✓ 使用低基数Tag
✓ 启用压缩传输
✓ 合理设置超时
✓ 实现重试机制

查询优化清单：

✓ 缩小时间范围
✓ 先过滤measurement和tag
✓ 使用窗口聚合
✓ 避免全表扫描
✓ 使用索引字段
✓ 实现查询缓存
✓ 监控慢查询
✓ 优化查询语句

存储优化清单：

✓ 合理设计Schema
✓ 使用低基数Tag
✓ 设置保留策略
✓ 实现降采样
✓ 使用SSD存储
✓ 监控磁盘空间
✓ 定期清理数据
✓ 压缩历史数据