简介基于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配有架构图和配置示例适合正在搭建日志平台或希望理解ELKSpark技术组合的工程师快速参考。目前已有109人学习下载。1. 日志处理平台从 ELK Stack 到 Spark Streaming 的实时链路三年前我接手过一套三百多台节点的数据仓库每天光 DB2 和 Tomcat 的日志就有几百 GB。那时候排查问题靠的是登到每台机器上 grep一条全链路报错查下来往往要耗掉半天。后来我把日志处理平台从零搭起来核心就是 ELK Stack 加 Kafka 和 Spark StreamingLogstash 负责采集解析、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 输出里会带着大量转义的\nES 索引体积和可读性都受影响。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就没必要硬换成 LogstashFlume 那边产生的数据照样可以进 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 HadoopES-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性能和可靠性都比第一种方式好。MapString, 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); SetString topics Collections.singleton(db2-log); JavaInputDStreamString messages KafkaUtils.createDirectStream( jssc, String.class, String.class, StringDecoder.class, StringDecoder.class, kafkaParams, topics ); messages.foreachRDD(rdd - { rdd.foreachPartition(partition - { while (partition.hasNext()) { Tuple2String, 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 rejectedES 节点 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 确认解析逻辑没有成为瓶颈。顺序反了容易互相甩锅日志平台排障的路径本来就不该是玄学一步步量出来最有底。希望帮到你。本文还有配套的精品资源点击获取