第3章数据存储与处理

3、数据存储与处理：Elasticsearch存储、时序数据库InfluxDB、流处理引擎Flink基础

好，咱们进入第三讲。这一章我打算聊聊智能运维系统的“数据底盘”。

你想想看，运维系统每天要处理多少数据？机器日志、监控指标、调用链追踪……这些数据格式不同，时效性要求也不同。有的要全文检索，有的要按时间聚合，有的要实时计算。一套方案打天下？不存在的。

我个人习惯，是把数据分三层来处理：存储层、时序层、流处理层。对应到具体工具，就是Elasticsearch、InfluxDB和Flink。今天咱们就把这三兄弟掰开揉碎了讲清楚。

3.1 Elasticsearch：日志与全文检索的基石

先说说Elasticsearch，简称ES。我在项目中遇到过很多次，团队把ES当成“万能存储”，什么数据都往里塞。结果呢？集群压力大，查询慢，还容易丢数据。ES最擅长的，其实是全文检索和日志分析。

ES的核心概念，说白了就几个：

索引（Index）：相当于数据库里的表。我习惯按天建索引，比如nginx-log-2025-04-01，方便管理。
文档（Document）：一条记录，JSON格式。比如一条Nginx访问日志。
分片（Shard）：数据切分到不同节点，实现分布式。
副本（Replica）：数据备份，保证高可用。

重要：ES的写入性能瓶颈通常在磁盘I/O。我建议用SSD，并且合理控制分片数量。一个分片建议20-40GB，不要超过50GB。

来看一个典型的日志写入示例：

# 创建一个索引，设置3个分片，2个副本
PUT /nginx-logs-2025.04.01
{
  "settings": {
    "number_of_shards": 3,
    "number_of_replicas": 2
  },
  "mappings": {
    "properties": {
      "@timestamp": { "type": "date" },
      "client_ip": { "type": "ip" },
      "request": { "type": "text" },
      "status": { "type": "integer" },
      "body_bytes": { "type": "long" }
    }
  }
}

# 写入一条日志
POST /nginx-logs-2025.04.01/_doc
{
  "@timestamp": "2025-04-01T10:15:30Z",
  "client_ip": "192.168.1.100",
  "request": "GET /api/v1/users",
  "status": 200,
  "body_bytes": 1234
}

查询的时候，ES的DSL语法很强大。比如查最近5分钟所有5xx错误：

GET /nginx-logs-*/_search
{
  "query": {
    "bool": {
      "filter": [
        { "range": { "@timestamp": { "gte": "now-5m" } } },
        { "range": { "status": { "gte": 500 } } }
      ]
    }
  },
  "sort": [ { "@timestamp": "desc" } ]
}

避坑指南：我曾经把ES的refresh_interval设成1秒，结果写入压力一大，集群直接打嗝。后来改成30秒，写入性能提升了3倍。记住，ES是近实时搜索，不是实时搜索。

3.2 InfluxDB：时序数据的专业选手

接下来是InfluxDB。这东西是专门为时序数据设计的。CPU使用率、内存占用、网络流量……这些数据的特点是：写多读少、按时间排序、定期聚合。

InfluxDB的数据模型，我简单说一下：

Measurement：类似表名，比如cpu_usage。
Tag：标签，会被索引。比如host=web01、region=cn-beijing。
Field：字段，实际数值。比如value=85.3。
Timestamp：时间戳，精度到纳秒。

为什么Tag和Field要分开？因为Tag会被索引，适合做筛选条件；Field不会被索引，适合做聚合计算。嗯，这里要注意：Tag不要太多，否则索引会膨胀。我一般控制在10个以内。

写入数据的语法很简单：

# 写入一条CPU使用率数据
INSERT cpu_usage,host=web01,region=cn-beijing value=85.3 1743480000000000000

查询用类SQL的语法，比如查最近1小时每台机器的平均CPU：

SELECT mean("value") 
FROM "cpu_usage" 
WHERE time > now() - 1h 
GROUP BY "host"

重要：InfluxDB的保留策略（Retention Policy）一定要设置。我见过有人不设，结果磁盘被撑爆。比如原始数据保留7天，然后自动降采样到1小时粒度，保留30天。

# 创建保留策略：原始数据保留7天
CREATE RETENTION POLICY "raw_7d" ON "mydb" DURATION 7d REPLICATION 1 DEFAULT

# 创建连续查询：每小时聚合一次，存入新表
CREATE CONTINUOUS QUERY "cq_1h" ON "mydb" 
BEGIN
  SELECT mean("value") INTO "cpu_usage_1h" 
  FROM "cpu_usage" 
  GROUP BY time(1h), "host"
END

注意：InfluxDB 1.x和2.x的语法差异很大。如果你还在用1.x，建议尽快迁移。2.x的Flux查询语言更强大，但学习曲线也陡一些。

3.3 Flink：流处理引擎的实时计算

最后是Flink。这东西是真正的实时计算引擎。ES和InfluxDB主要负责存储和查询，而Flink负责在数据流动的过程中做计算。

我举个例子你就明白了。假设我们要做实时异常检测：每秒钟有10万条CPU指标进来，我们要在5秒内判断哪些机器的CPU突然飙升，然后触发告警。这事ES和InfluxDB干不了，因为它们都是“存下来再查”。Flink可以做到数据一来就处理。

Flink的核心概念：

Source：数据源，比如Kafka、Socket。
Transformation：转换操作，比如map、filter、window。
Sink：输出目标，比如ES、InfluxDB、告警系统。
Window：窗口，比如5秒的滑动窗口。

来看一个简单的Flink作业，从Kafka读取CPU指标，计算每台机器最近1分钟的平均值，然后写入InfluxDB：

// 伪代码，展示核心逻辑
DataStream<CpuMetric> stream = env
    .addSource(new FlinkKafkaConsumer<>("cpu-topic", ...))
    .keyBy(metric -> metric.getHost())
    .window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
    .apply(new WindowFunction<CpuMetric, AvgCpu, String, TimeWindow>() {
        @Override
        public void apply(String host, TimeWindow window, 
                          Iterable<CpuMetric> input, Collector<AvgCpu> out) {
            double sum = 0;
            long count = 0;
            for (CpuMetric m : input) {
                sum += m.getValue();
                count++;
            }
            out.collect(new AvgCpu(host, sum / count, window.getEnd()));
        }
    });

// 写入InfluxDB
stream.addSink(new InfluxDbSink<>("http://influxdb:8086", "mydb", "cpu_avg_1m"));

避坑指南：我曾经在Flink作业里用了ProcessingTime，结果因为机器时钟不同步，数据乱序严重。后来改成EventTime，配合Watermark机制，才解决了问题。记住：生产环境一定要用EventTime。

3.4 三者的协同：一张图看懂

这三兄弟不是孤立的。我画了一张图，展示它们在智能运维系统中的位置：

这张图展示了典型的数据流：

数据源（日志、指标）进入Kafka等消息队列。
Flink从Kafka消费数据，做实时计算（比如聚合、异常检测）。
计算结果写入ES（用于日志检索）和InfluxDB（用于时序分析）。
上层应用从ES和InfluxDB读取数据，展示在仪表盘或触发告警。

核心思路：ES负责“查”，InfluxDB负责“看趋势”，Flink负责“算”。各司其职，不要混用。

3.5 选型建议与总结

最后，我根据实际经验，给一些选型建议：

场景	推荐工具	原因
日志全文检索	Elasticsearch	强大的分词和搜索能力
监控指标存储	InfluxDB	高写入性能，自动降采样
实时异常检测	Flink	低延迟，精确一次语义
告警规则引擎	Flink + InfluxDB	Flink做计算，InfluxDB存历史

嗯，这一章的内容就到这里。记住一句话：没有最好的工具，只有最合适的组合。ES、InfluxDB、Flink这三兄弟，用好了就是智能运维的“三驾马车”。

公众号:蓝海资料掘金营，微信deep3321

📚 课程目录

3、数据存储与处理：Elasticsearch存储、时序数据库InfluxDB、流处理引擎Flink基础

3.1 Elasticsearch：日志与全文检索的基石

3.2 InfluxDB：时序数据的专业选手

3.3 Flink：流处理引擎的实时计算

3.4 三者的协同：一张图看懂

3.5 选型建议与总结