3. 数据采集层设计

数据采集层是 Elastic 批流数据管道的入口。它决定了后续索引、检索、聚合与告警能拿到什么样的原始素材。采集层设计要解决三个问题：从哪里采（数据源类型与位置）、怎么采（采集器与输入插件、解码器）、采成什么样（字段规范化与统一元数据）。Elastic 生态中常用的采集组件分为两类：轻量级的 Beats 家族（Filebeat、Metricbeat 等）负责在数据源侧就近采集并做初步结构化；Logstash 则承担更重的解析、转换、富化和多源汇聚职责。二者不是替代关系，而是分层协作关系。

采集组件分工与选型

Filebeat 面向日志与文本文件，核心能力是跟踪文件偏移量（registry）、多行合并、背压控制，资源占用低，适合部署在每台业务主机上。Metricbeat 面向系统与服务的指标，通过模块（module）和指标集（metricset）周期性地从操作系统、数据库、中间件、容器平台拉取数值型数据。Logstash 具备丰富的输入、过滤、输出插件，适合处理需要复杂解析、跨源关联、条件路由的场景。一个常见的分层是：主机侧用 Filebeat/Metricbeat 采集并做轻量字段处理，汇聚到 Logstash 做统一解码、字段规范化和路由，再写入 Elasticsearch。

选型判断可以遵循：纯文件日志且解析简单，优先 Filebeat；纯指标且来源有官方模块，优先 Metricbeat；需要 grok、dissect、ruby、多路输出、与外部系统交互，或需要把多种 Beats 数据统一治理，则引入 Logstash。若数据源是 Kafka、数据库 JDBC、HTTP 接口，Logstash 的对应输入插件通常比自研采集更省成本。

输入插件与解码器配置要点

Filebeat 的输入推荐使用 filestream 类型，它替代了旧的 log 输入，支持更可靠的文件状态跟踪。关键参数包括 paths（采集路径）、prospector.scanner、parsers（解码器链）以及 close.on_state_change。多行日志（如 Java 异常堆栈）必须配置 multiline，否则一条异常会被拆成多行独立事件，破坏可读性。下面是一个 Filebeat 采集应用日志并做 JSON 解码的示例配置：

filebeat.inputs:
  - type: filestream
    id: app-json-log
    paths:
      - /var/log/app/*.log
    parsers:
      - ndjson:
          target: ""
          add_error_key: true
          overwrite_keys: true
    fields:
      env: prod
      service: order-api
    fields_under_root: true

processors:
  - add_host_metadata: ~
  - add_cloud_metadata: ~
  - drop_fields:
      fields: ["agent.ephemeral_id", "ecs.version"]

这里 ndjson 解码器把每行 JSON 展开为顶层字段，fields_under_root: true 让自定义字段直接进入事件根层级，便于后续按 env、service 过滤。add_host_metadata 自动补充主机名、操作系统等信息，是字段规范化的重要一环。

Metricbeat 的配置以模块为单位，每个模块下可启用多个 metricset，并设置 period 控制采集频率。指标类数据要特别注意单位与量纲的统一，例如内存统一用字节、时间统一用毫秒，避免后续聚合时出现量级混乱。

metricbeat.modules:
  - module: system
    period: 10s
    metricsets: ["cpu", "memory", "filesystem", "network"]
    cpu.metrics: ["percentages", "normalized_percentages"]
    filesystem.ignore_types: ["tmpfs", "devtmpfs"]

  - module: mysql
    period: 30s
    metricsets: ["status", "galera_status"]
    hosts: ["root:${MYSQL_PWD}@tcp(127.0.0.1:3306)/"]

Logstash 解码与字段规范化

Logstash 的输入插件决定数据从哪来，codec 决定字节流如何切分为事件。常见组合是 beats 输入接收 Filebeat/Metricbeat 数据，kafka 输入消费消息队列，jdbc 输入周期性拉取库表。过滤阶段是字段规范化的主战场：grok 用于非结构化文本，dissect 在分隔符固定时性能更好，date 把业务时间写入 @timestamp，mutate 完成重命名、类型转换、删除冗余字段。

input {
  beats { port => 5044 }
  kafka {
    bootstrap_servers => "kafka1:9092"
    topics => ["order-events"]
    codec => json
    group_id => "logstash-order"
  }
}

filter {
  if [service] == "order-api" {
    grok {
      match => { "message" => "%{TIMESTAMP_ISO8601:log_time} %{LOGLEVEL:level} %{GREEDYDATA:msg}" }
    }
    date {
      match => ["log_time", "ISO8601"]
      target => "@timestamp"
      timezone => "Asia/Shanghai"
    }
    mutate {
      rename => { "msg" => "message" }
      convert => { "level" => "string" }
      remove_field => ["log_time", "host.os.version"]
    }
  }
}

output {
  elasticsearch {
    hosts => ["https://es1:9200"]
    data_stream => "logs-order-api-default"
    user => "ingest_writer"
    password => "${ES_PWD}"
  }
}

这段配置体现了规范化的核心动作：把业务时间解析为统一时区的 @timestamp，把日志级别、消息体归入标准字段名，删除无检索价值的冗余字段。注意 date 插件必须显式声明 timezone，否则 Logstash 会按运行环境时区解释，导致跨区域数据时间错位。

多源采集的常见问题与判断依据

采集层最容易出问题的地方有三类。第一类是重复与丢失：Filebeat 依赖 registry 记录偏移，若 registry 被误删或容器重建未持久化，会从头重读造成重复；Logstash 的 beats 输入在未开启持久化队列时，进程崩溃可能丢事件。判断依据是比对源文件行数与 Elasticsearch 文档数，并检查 filebeat.registry 是否挂载到持久卷。第二类是解析失败：grok 不匹配时会产生 _grokparsefailure 标签，可通过查询带该标签的文档量来定位。第三类是时间错乱：表现为文档集中在采集时刻而非业务发生时刻，通常是没有配置 date 插件或时区设置错误。

一个可执行的验证方法是：先在单台主机上以调试模式运行采集器，观察输出事件结构，确认字段名、类型、时间都符合预期后，再批量下发配置。对于 Logstash，可用 --config.test_and_exit 校验语法，用 stdout { codec => rubydebug } 临时输出到控制台检查事件形态。只有采集层字段稳定、时间准确、无大量解析失败，后续的索引生命周期管理、聚合分析和告警才有可靠基础。