简介：基于ELK Stack与Spark Streaming的日志处理平台是一份面向大数据运维与开发人员的PDF文档，围绕日志从采集、解析、存储到实时分析与可视化的完整链路展开。内容梳理了日志处理方案从v1.0到v3.0的演进，对比离线批处理与实时搜索差异，重点讲解Logstash的input、filter、output三阶段插件机制，并以DB2多行日志为例演示multiline codec、grok、mutate和date插件配合使用；随后阐述Elasticsearch的索引、集群与近实时搜索原理，以及Kibana仪表盘如何动态监测数据库实例状态。文档还介绍了Kafka在数据管道中的角色，以及Spark Streaming与ELK Stack、Hadoop深度整合，构建支持高并发流式处理与离线批处理的统一日志分析系统。PDF共1个文件，大小1.49MB，配有架构图和配置示例，适合正在搭建日志平台或希望理解ELK+Spark技术组合的工程师快速参考。目前已有109人学习下载。
1. 日志处理平台：从 ELK Stack 到 Spark Streaming 的实时链路
三年前我接手过一套三百多台节点的数据仓库，每天光 DB2 和 Tomcat 的日志就有几百 GB。那时候排查问题靠的是登到每台机器上 grep，一条全链路报错查下来往往要耗掉半天。后来我把日志处理平台从零搭起来，核心就是 ELK Stack 加 Kafka 和 Spark Streaming：Logstash 负责采集解析、Kafka 做缓冲削峰、Elasticsearch 提供近实时检索、Kibana 出仪表盘、Spark Streaming 跑实时统计和告警。这套方案最大的价值就是把日志从“事后追查的黑匣子”变成了“实时可观测的流水线”，对运维排障、业务分析和异常告警都有直接帮助。这篇笔记面向正在选型或准备动手自建日志平台的工程师，我把每个组件的选型理由、关键参数和踩过的坑都写在下面。
2. Logstash 采集与解析：一套配置吃透多行日志和字段提取
2.1 为什么用 Logstash 而不是 Flume
日志处理链路的起点是采集。Flume 吞吐量更大、与 HDFS 整合更好，七牛那种每天 500 亿条、峰值 80 万 TPS 的场景选 Flume 是合理的。但大多数企业的日志量在每天 TB 级别，Logstash 的解析能力才是真正的优势。它内置了几十种 input、filter、output 插件，配置一次就能完成采集、过滤、字段提取、格式转换整条链路，而 Flume 做这些事需要写很多自定义代码。
对我来说选择 Logstash 还有一个实际原因：团队里没人愿意为采集端单独维护一套 Java 代码，Logstash 的纯配置方式让运维同事也能上手改。从社区活跃度和插件丰富度来看，Logstash 也在持续迭代，Elastic 官方的支持力度明显更大。
2.2 多行日志怎么合并：multiline codec 的配置细节
实际生产环境里最头疼的不是单行日志，而是像 DB2 那样一条日志跨几十行甚至几百行的半结构化文本。每一条 DB2 日志以“2014-10-19-12.19.46.033037-300”格式的时间戳开头，后续的行都是这条日志的内容。如果不做合并，Elasticsearch 里存的就会是碎片，搜索和分析根本没有意义。
input { file { path => "/data/db2/logs/instance1/*.log" codec => multiline { pattern => "^\\d{4}-\\d{2}-\\d{2}-\\d{2}\\.\\d{2}\\.\\d{2}\\.\\d{6}[\\+-]\\d{3}" negate => true what => previous max_lines => 1000 } } }这段配置里最重要的三个参数：pattern用来匹配日志条目的起始行，negate => true表示不匹配该模式的行都归到上一条日志里，what => previous指定把当前行追加到前一条消息的尾部。max_lines是很多人忽略的坑，默认值在 DataStage 那种超长日志场景下根本不够用，一条日志超过 500 行直接报错或截断。我一般会设到 1000 以上，同时配合max_bytes限制单条消息的体积，防止一条日志把内存撑爆。
2.3 日志字段提取：grok、mutate、date 三件套
合并完成的多行日志还是一个整体文本，Kibana 上做聚合分析的前提是把时间戳、日志级别、模块名、实例名这些字段从文本里拆出来。拆字段的主力是 grok，它是 Logstash 内置的正则解析插件，通过预定义模式加自定义正则完成匹配。
filter { mutate { gsub => ["message", "\\n", " "] } grok { match => { "message" => "%{TIMESTAMP_ISO8601:log_time} %{WORD:level} %{GREEDYDATA:msg_body}" } remove_field => ["message"] } date { match => ["log_time", "yyyy-MM-dd-HH.mm.ss.SSSSSS"] target => "@timestamp" } }这里的mutate插件把多行合并时保留的换行符替换成空格，否则 JSON 输出里会带着大量转义的\n，ES 索引体积和可读性都受影响。grok的匹配规则里TIMESTAMP_ISO8601是内置时间戳模式，WORD匹配单词，GREEDYDATA匹配剩余所有字符，捕获到的内容分别命名为 log_time、level、msg_body。date插件解析日志原始时间戳并覆盖到@timestamp，这一步很关键——如果 Logstash 采集有延迟，默认的@timestamp是处理时间而不是日志产生时间，排序和聚合都会出错。
grok 匹配不上时配置不会报错，但字段就提取不出来，所以在测试阶段建议先用rubydebug编码器输出实际解析结果确认一下。
3. Kafka 缓冲层：日志不直接进 Elasticsearch 的三个理由
3.1 削峰填谷和数据共享
很多刚刚接触 ELK 的团队会把 Logstash 的 output 直接指到 Elasticsearch，这个架构在数据量小时没问题，但一旦业务高峰和日志洪峰重叠，ES 的写入压力会剧烈波动。我在中间加了一层 Kafka，原因有三个：第一是削峰填谷，Logstash 生产的速率和 ES 消费的速率可以解耦，即使 Logstash 突然涌入大量日志也不至于把 ES 打挂；第二是数据共享，日志进 Kafka 后可以同时被 Spark Streaming、离线 HDFS 任务和 Kibana 消费，不需要为每个下游单独采集一遍；第三是故障隔离，ES 集群重启或 rollover 时 Kafka 会积压消息，恢复后自动补发，日志不会丢。
3.2 主题划分和分区策略
Kafka 里主题的划分直接决定了下游消费的便捷性。我一般按照日志产生方来建主题，DB2 一个主题、Tomcat 一个主题、应用业务日志单独一个主题，每个主题再按实例或分片设置分区数。
# Logstash 作为 Kafka 生产者 output { kafka { bootstrap_servers => "kafka1:9092,kafka2:9092,kafka3:9092" topic_id => "db2-log" codec => json batch_size => 16384 linger_ms => 50 } }生产端的核心参数是batch_size和linger_ms。batch_size控制单批次消息大小，linger_ms是消息在内存里等待打包的时间窗口，设得太小吞吐上不去，设得太大延迟会变高。我通常把linger_ms设在 20 到 100 毫秒之间，让 Logstash 攒够一个小批量再发给 Kafka，吞吐和延迟都能兼顾。
3.3 WAL 机制为什么能让 Kafka 既可靠又快
Kafka 保证可靠性的关键是把所有消息按顺序写入磁盘上的 Write Ahead Log，也就是 WAL。它不会因为落盘就变慢，实时数据传输路径仍然走内存，磁盘写入只是异步持久化。另一方面 ZooKeeper 保存了每个消费者组在每个分区上的偏移量，消费者崩溃后重启可以直接从上次的位置继续消费，不会重复也不会丢。
# 查看消费者组当前偏移量 kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --group spark-streaming-app --describe每过一段时间我都会执行一次这个命令，看看LOG-END-OFFSET和CURRENT-OFFSET的差距。如果 lag 持续增长，说明下游 Spark Streaming 计算速度跟不上，需要调整批处理间隔或增加分区数。
3.4 Flume 与 Logstash 的选型边界
Kafka 这层选完了，采集端仍存在 Flume 和 Logstash 的选择问题。单纯从吞吐量来看 Flume 占优，它和 HDFS 的整合也更成熟。如果日志源本身是已经接入 Flume 的 Flume Agent，就没必要硬换成 Logstash，Flume 那边产生的数据照样可以进 Kafka。但如果下游需要对日志做复杂的格式清洗、字段提取和多行合并，Logstash 的 filter 插件效率高得多。这里我的做法是具体场景具体分析：采集端吞吐瓶颈明显的选 Flume，解析清洗逻辑复杂的选 Logstash，量又大解析又复杂的就在 Flume 之后再加一层 Logstash。
4. Elasticsearch 检索集群：从索引设计到脑裂防御
4.1 文档、索引、类型与关系型数据库的类比
Elasticsearch 存储和检索的基础概念对应关系型数据库比较容易理解：Index 对应数据库实例，Type 对应表，Document 对应行。日志数据经过 Logstash 处理后会格式化成 JSON 文档，再由 Elasticsearch output 插件自动写入。
output { elasticsearch { hosts => ["es1:9200", "es2:9200", "es3:9200"] index => "db2-log-%{+YYYY.MM.dd}" document_type => "_doc" } }索引名我习惯按天切分，每天自动生成一个新的索引文件，到了清理时间直接删掉过期索引释放磁盘，比在 ES 内部逐条删数据高效得多。document_type在新版本里已经弱化，统一用_doc即可。
4.2 ES-Hadoop：把离线计算和实时搜索打通
传统 Hadoop 生态里，离线批处理结果落到 HDFS 后就很难再做实时查询。Elasticsearch for Apache Hadoop（ES-Hadoop）这个组件弥补了这一点：Hadoop 的计算结果可以直接写回 ES 索引，供 Kibana 和业务系统实时检索；反过来 ES 中积累的日志数据也可以作为 Spark 或 Hive 的数据源做更复杂的分析。ES-Hadoop 的架构打通了离线批处理和在线检索之间的壁垒，不需要再维护一套数据同步任务。
4.3 集群自动发现机制的坑：脑裂
Elasticsearch 默认基于 Gossip 路由协议自动发现集群节点，只要cluster.name配置一致，节点之间会自动组队。听起来很方便，但实际运行中这种自动发现并不可靠，经常出现节点间网络抖动导致部分节点互相看不到对方，最后集群分裂成多个小集群，也就是所谓的脑裂。脑裂一旦发生，同一个索引的多个分片会在不同集群里被同时写入，数据一致性和可用性都受到严重威胁。
我的做法是完全关闭自动发现，在每个节点上显式列出其他节点的 IP 或主机名，让节点在启动时进行被动发现。同时把minimum_master_nodes设置成(节点数 / 2) + 1，防止少数派节点在那个配置下强行选主，这一项配置在防止脑裂上非常关键。
4.4 新浪的调优思路延伸
新浪的日志团队每天处理约 32 亿条、约 2TB 的日志，他们的经验中有两个特别值得借鉴的点：一是调整max open files，日志量大的节点默认文件句柄数远远不够，ES 运行一段时间后就会出现 “Too many open files” 的告警；二是自研索引管理系统来统一处理索引的创建、优化和删除，日志平台索引数量多起来之后人工维护是不现实的。另外中文场景下需要给 ES 安装elasticsearch-analysis-ik分词插件，否则中文搜索基本不可用。
5. Spark Streaming 实时计算：两种 Kafka 消费方式对比与实战
5.1 Spark Streaming 与 Storm 的本质差异
Storm 是事件级别的流处理框架，来一条处理一条，端到端延迟通常在毫秒级；Spark Streaming 的模型是 mini-batch，把连续的数据流按时间窗口切成一个个微批次，再交给 Spark 引擎做批处理。这个差异决定了选型方向：对延迟极度敏感、需要逐条处理的状态计算选 Storm，对吞吐量要求高、能够容忍秒级延迟的日志统计和告警场景选 Spark Streaming 性价比更高。日志处理平台里绝大多数场景属于后者，所以我选 Spark Streaming。
5.2 基于高级 API 的消费方式与 WAL 的引入
Spark Streaming 从 Kafka 读数据有两种方式。第一种是基于 Kafka 消费者高级 API，在 Spark 工作节点的执行器上创建消费者线程直接订阅。这种方式配置简单，但存在一个致命问题：Spark Job 失败时，已经从 Kafka 拉取到执行器但还没处理完的数据会丢失。官方给出的补救方案是在 Spark Streaming 里开启 Write Ahead Logs，把收到的数据先写到 HDFS 上再处理。
sparkConf.set("spark.streaming.receiver.writeAheadLog.enable", "true");开启这个参数后执行器消费到的数据会同步落一份到分布式文件系统，Spark 通过 WAL 恢复数据。但这种方案有额外的落盘开销，而且偏移量的维护仍然在 ZooKeeper 上，程序崩溃时可能出现重复消费或数据不一致。
5.3 createDirectStream 直连模式为什么更稳
第二种方式是我现在首选的做法：用createDirectStream接口直接连接 Kafka broker 读 WAL，不创建消费者线程。Kafka 的分区和 Spark 的 RDD 分区是一对一的映射关系，每个 Spark 分区自己管理对应的 Kafka 分区偏移量，不需要再维护额外的 WAL，性能和可靠性都比第一种方式好。
Map<String, String> kafkaParams = new HashMap<>(); kafkaParams.put("bootstrap.servers", "kafka1:9092,kafka2:9092"); kafkaParams.put("group.id", "spark-streaming-app"); kafkaParams.put("auto.offset.reset", "largest"); Set<String> topics = Collections.singleton("db2-log"); JavaInputDStream<String> messages = KafkaUtils.createDirectStream( jssc, String.class, String.class, StringDecoder.class, StringDecoder.class, kafkaParams, topics ); messages.foreachRDD(rdd -> { rdd.foreachPartition(partition -> { while (partition.hasNext()) { Tuple2<String, String> record = partition.next(); JSONObject json = new JSONObject(record._2); String level = json.getString("level"); if ("ERROR".equals(level)) { // 发送告警或写入统计结果 } } }); });这段代码里kafkaParams中比较核心的还有auto.offset.reset，它决定了应用在找不到偏移量时从哪里开始读，largest表示从最新消息开始，smallest表示从最早的消息开始。日志平台建议用largest，如果是需要回放历史数据做补算，再改成smallest。createDirectStream的偏移量完全由 Spark Streaming 通过检查点机制自己维护，程序重启后从检查点恢复偏移量，既不会重复读也不会漏读，消除了和 ZooKeeper 之间的数据不一致风险。
5.4 日志告警实践：把异常变成邮件通知
拿到解析后的 JSON 日志后，Spark Streaming 能做的事情就多了。针对 DB2 实例如果检测到级别为 ERROR 的日志，可以触发邮件告警发送到运维邮箱。我这里实际做过的做法是先在foreachRDD里过滤出 ERROR 级别的日志，再按数据库实例聚合，统计一分钟内出错次数，超过阈值就调用邮件接口发送摘要信息。阈值不要写死在代码里，建议放到外部配置或 Redis 中动态调整，否则每次改阈值都要重新打包提交任务。
6. 日志平台常见问题排查与避坑记录
6.1 multiline 把独立日志吞成一条
现象：Elasticsearch 里出现体积特别大的文档，一条日志里包含了几十条不相关的内容，而且@timestamp是第一条日志的时间。 原因：multiline的pattern没有匹配到新日志的起始行，导致所有新日志都被归到上一条里。常见于日志格式发生过调整，比如时间戳格式从“2014-10-19-12.19.46”变成了“2019-10-19 12:19:46”，原有正则失效。 解决：修改pattern使其覆盖所有可能的时间戳格式，并在 Logstash 配置热加载后用一条真实日志做验证，确认消息拆分正确后再放量。
6.2 Elasticsearch 节点频繁掉线后又自动回来
现象：ES 集群状态在黄色和红色之间波动，部分节点日志里出现 “master not discovered” 的报错。 原因：节点数超过 3 个以后，默认的 Gossip 自动发现机制在网段隔离或交换机抖动时变得不可靠，少数派节点无法找到主节点，出现脑裂或节点反复加入退出。 解决：在elasticsearch.yml里显式配置discovery.zen.ping.unicast.hosts列出所有节点地址，并设置minimum_master_nodes为法定节点数量，这样即使有节点临时失联也不会触发错误的主节点选举。
6.3 Spark Streaming 重启后重复消费
现象：日志告警应用每次重启后都会把过去几分钟的日志重新跑一遍，重复发送告警邮件。 原因：使用的是基于高级 API 的消费方式，偏移量虽然保存在 ZooKeeper 中，但 Spark Streaming 检查点和 ZooKeeper 的偏移量并不完全同步，程序异常退出时两者不一致。 解决：切换到createDirectStream直连模式，让偏移量由 Spark 的检查点机制统一管理。注意检查点在第一次正常启动后才会生成，首次运行别急着 kill 进程。
6.4 Elasticsearch 写入速度上不去
现象：日志量峰值时 Logstash 频繁报bulk request rejected，ES 节点 CPU 高企但写入 TPS 不高。 原因：ES 默认的分片数和副本数配置在日志场景下不理想，单个分片写入瓶颈限制了整体吞吐；同时refresh_interval默认的 1 秒刷新间隔导致每个写入请求都触发代价较高的 refresh 操作。 解决：按天建立索引并控制单索引分片数，副本数在日志场景下可以设为 1。如果延迟允许，把refresh_interval调大到 30 秒甚至 60 秒，写入性能会有明显提升。
6.5 Kafka 消费者组 lag 持续增长
现象：消费者组在kafka-consumer-groups.sh里看到的 lag 越来越大，实时性逐渐变差。 原因：下游消费能力跟不上生产速率，最常见的是 Spark Streaming 的批处理间隔设得太短，每一批还没算完下一批就到了，任务不断积压。 解决：先把批处理间隔调大观察 lag 趋势，同时增加 RDD 分区数来提升并行度，如果还不能解决，回到上游限制 Logstash 的生产速率或者增加 Kafka 主题分区数。
6.6 Logstash 运行状态成黑匣子
现象：Logstash 进程还在，但日志不再写入 ES，内部 state 信息只能靠猜。 原因：Logstash 没有像 ES 那样内置完善的监控接口，端口还在监听但处理线程已经卡死的情况并不少见。 解决：在配置里开启xpack.monitoring把 Logstash 指标上报到 ES，同时额外写一个定时任务检查 Logstash 处理的最新时间戳，如果超过 N 分钟没有新消息及时告警。
7. 进阶技巧：Kafka 主题分区与 Spark 并行度匹配调优
日志平台上线初期，我遇到的第一个性能瓶颈就是数据从 Kafka 到 Spark 的摄入速率上不去。排查下来发现 Kafka 主题建了 8 个分区，而 Spark Streaming 默认只给对应的 RDD 分配了 2 个分区，等于只有 2 个计算任务在跑，剩下 6 个分区的数据全部积压。
Kafka 分区和 Spark RDD 分区是createDirectStream直连模式下的一对一关系，Kafka 有多少个分区，RDD 就会建多少个分区。理想情况下两者相等，并行度已经天然匹配。但实际生产中下游计算逻辑如果比较复杂，单个分区处理速度跟不上拉取速度，还是需要增加 Kafka 分区数来提升并行度。我给一组参考数值：单条日志平均 2KB、单分区消费速率能达到每秒 5000 条的情况下，每天 2TB 日志量建议建 24 到 32 个分区，Spark 分配的 executors 至少保证 4 个以上，每 executor 的核数在 2 到 4 之间。
另一个容易被忽略的参数是spark.streaming.kafka.maxRatePerPartition，它限制每个分区每秒最多拉取多少条记录。这个参数在 Kafka 生产速率远大于 Spark 处理能力时能起到限流保护作用，防止数据在 Spark 执行器里堆积导致 OOM。具体值需要根据单条日志大小和计算逻辑实测，我一般从 10000 起调，观察批次处理耗时接近批处理间隔的 70% 到 80% 时为佳。
从那以后我每次调整日志链路参数，都强制走一遍同样的流程：先看 Kafka 的生产消费速率是否平衡，再调 Spark 的批处理间隔和并行度，最后回到 Logstash 确认解析逻辑没有成为瓶颈。顺序反了容易互相甩锅，日志平台排障的路径本来就不该是玄学，一步步量出来最有底。希望帮到你。
本文还有配套的精品资源，点击获取
