2. Elastic数据管道基础

在 Elastic 生态中，数据管道并不是单一软件，而是一组职责明确、可组合的组件协同工作的结果。理解 Elasticsearch、Logstash、Beats、Kibana 四者的角色定位，是设计批流一体管道的前提。它们分别承担存储与检索、转换与路由、采集与轻量处理、可视化与探索四类职责，彼此之间通过 HTTP/REST、Beats 协议或 Kafka 等中间层连接。

四个组件的角色定位

Elasticsearch 是管道的终点与核心存储。它负责索引、分片、副本、倒排索引与聚合计算，是数据最终可被检索和分析的地方。在批处理场景中，它接收批量写入的索引请求；在流处理场景中，它通过 _bulk 接口持续接收小批量文档。它的写入性能、刷新间隔（refresh_interval）和分片策略直接决定管道的吞吐与延迟。

Logstash 是管道的“转换与路由中枢”。它采用 input → filter → output 三段式结构，擅长解析非结构化文本、做字段提取、类型转换、条件路由和多目标输出。它适合处理复杂转换，但资源开销较大，通常部署在独立节点上，而不是与 Elasticsearch 混部。

Beats 是轻量级采集器家族，包括 Filebeat、Metricbeat、Packetbeat 等。它们部署在数据源附近，负责把日志、指标、网络流量等原始数据采集并发送到 Logstash 或 Elasticsearch。Beats 只做轻量处理（如多行合并、字段添加），不做复杂转换，因此资源占用低、适合边缘部署。

Kibana 是管道的“观察窗口”。它不参与数据搬运，但通过 Discover、Visualize、Dashboard 和 Lens 让用户验证管道结果：文档是否按预期字段解析、时间戳是否正确、聚合指标是否符合业务预期。没有 Kibana 的验证，管道是否真正可用就无从判断。

批处理与流处理的边界

批处理与流处理的边界，本质上不是“数据量大小”，而是触发方式与延迟容忍度。批处理由调度器按固定周期触发，一次处理一批有界数据，典型延迟在分钟到小时级；流处理由事件到达触发，持续处理无界数据，典型延迟在秒级甚至亚秒级。

在 Elastic 管道中，这条边界体现为三条具体差异：

• 写入方式：批处理使用大尺寸 _bulk 请求（例如每批 5000～10000 条文档），流处理使用小批量高频写入（例如每批 500～1000 条，间隔 1～5 秒）。
• 刷新策略：批处理可临时调大 refresh_interval（如 30s）以提升吞吐，写入完成后再恢复；流处理需要保持较低刷新间隔（如 1s）以保证可查询性。
• 错误处理：批处理失败可整批重跑，幂等性容易保证；流处理需要处理部分失败、重复消费和乱序问题，通常依赖文档 ID 去重或 _version 控制。

协同方式：批流一体的三种典型模式

实际生产中，批与流并非互斥，而是按数据时效性分层协同。常见模式有三种：

模式一：流为主、批补历史。 实时链路用 Beats → Logstash → Elasticsearch 持续写入当天数据；历史归档数据在夜间用 Logstash 批量导入。两者写入同一索引或按日期分索引（如 logs-2025.06.01），Kibana 通过索引模式统一查询。

模式二：批为主、流补增量。 核心数据仓库每天批量同步一次，但关键业务指标通过流处理实时更新到独立索引，Kibana 同时展示 T+1 全量报表和实时看板。

模式三：Kafka 作为批流统一缓冲层。 Beats 或应用先把数据写入 Kafka，Logstash 从 Kafka 消费后写入 Elasticsearch。批处理任务也从 Kafka 读取同一份数据做离线计算。这样批与流共享同一数据源，避免双写不一致。

一个可参考的管道配置片段

下面是一个 Logstash 同时服务批流两种输入的简化配置示例，展示如何用同一套 filter 处理不同来源：

input {
  beats { port => 5044 }                    # 流：Beats 实时推送
  file {
    path => "/data/archive/*.json"          # 批：历史文件
    start_position => "beginning"
    sincedb_path => "/dev/null"             # 示例：每次重跑
    codec => "json"
  }
}

filter {
  date {
    match => ["event_time", "ISO8601"]      # 统一时间字段
    target => "@timestamp"
  }
  mutate {
    remove_field => ["event_time", "host"]
  }
}

output {
  elasticsearch {
    hosts => ["http://es-node:9200"]
    index => "logs-%{+YYYY.MM.dd}"
    document_id => "%{[event_id]}"          # 幂等去重
  }
}

这段配置的关键点在于：document_id 使用业务事件 ID，使批处理重跑和流处理重复消费都能覆盖同一文档，而不是产生重复记录。这是批流协同中最实用的幂等手段之一。

如何判断该用批还是流

选择批或流，可以用三个问题快速判断：数据是否必须“到达即可见”？下游是否容忍分钟级延迟？失败后能否接受整批重跑？如果三个答案分别是“是、否、否”，优先流处理；如果分别是“否、是、是”，批处理更简单可靠。多数 Elastic 管道是两者混合：实时告警走流，日报报表走批，Kibana 统一呈现。

理解四组件角色与批流边界后，下一步才是具体配置与调优。本节的重点是建立“谁负责什么、数据从哪来到哪去、批与流在哪些点上分叉又在哪里汇合”的整体认知，这比记住某个参数更重要。