简介：基于Spark的实时日志分析及异常检测系统，是一份面向计算机、电子信息工程、数学等专业学生课程设计、期末大作业和毕业设计的完整工程源码包。项目整合Flume、Kafka、HBase、Spark Streaming与Scala技术栈，覆盖日志采集、消息缓冲、分布式存储、实时流处理及异常检测核心链路；代码采用参数化编程、注释明细，内含运行结果，便于调试与二次开发。压缩包共有14个文件，以Scala源文件、Eclipse及IDEA工程配置xml、编译输出class文件为主，附带README.md说明文档和kotlin_module配置，整体约18KB，目录结构清晰，可快速导入开发环境。目前已有148人学习下载。作者为资深算法工程师，长期从事大数据与AI仿真，源码经测试运行成功，对希望掌握完整实时日志处理链路、快速复现异常检测流程的读者，具有明确参考价值。
1. 实时日志分析做到什么程度才算能用：这套Spark系统解决什么、适合谁
值班最大的痛点不是服务器宕机，而是日志量翻了三倍却不知道哪里先出问题。中午订单接口耗时从80毫秒爬到800毫秒，业务群里问了一圈，值班同事才在ELK里翻到一条异常堆栈——这时已经过去了二十分钟。基于Spark的实时日志分析及异常检测系统，核心就是把“事后翻日志”变成“指标曲线刚抬头就通知你”，配套源代码和文档说明，从Kafka接入、Structured Streaming清洗聚合，到阈值异常检测和告警输出，一条流水线完整跑通。
这套方案适合日志量单日亿级以下、延迟目标在秒级到分钟级的场景，也适合刚完成Spark集群搭建、想找一个完整数据管道做落地的团队。很多入门项目只做到“把日志打印到控制台”就结束了，离可用还差很远：没有checkpoint、没有窗口聚合、没有异常判定，跑一个晚上就内存溢出。这篇笔记直接讲生产可用的工程骨架，覆盖架构分工、可运行的代码、参数怎么设和几类常见的翻车现场。
2. 管道先于算法：Spark实时日志分析的系统架构与选型理由
2.1 为什么是Spark而非自己写Kafka消费者
很多团队接到“实时日志分析”需求，第一反应是自己写一个Kafka消费者：循环poll、解析JSON、攒批写数据库。五百行代码能跑，但上线之后问题集中爆发：消费者组重平衡时offset怎么处理、任务异常退出从哪恢复、窗口聚合的状态要自己维护、多个消费者实例怎么分担数据压力。用Spark Structured Streaming，offset管理、检查点、状态存储、故障恢复都交给框架，你实际只需要写transform和foreachBatch两段逻辑。这是把系统建立在Spark上最核心的理由——调度和恢复机制框架已经消化了，自己从头做到同样的可靠程度成本太高。
选型上还有一个务实原因。多数团队不是没有Spark，而是已经有一个在跑离线ETL的Spark集群；把实时分析挂到集群上，资源复用，运维习惯和监控指标也统一。Flink在低延迟流处理上确实更激进，但日志类分析延迟目标通常在秒级到分钟级，Spark的微批模型完全够用；全链路开发调试资料多，新手团队两三天能上手，而Flink的状态管理和反压机制要适应需要更长周期。这里要提醒一句：如果业务要求事件发生后1秒内必须触发动作，Spark不是合适选项，选型文档里要先把延迟前提写清楚。
2.2 整体数据流与各层职责
链路是常见做法：应用日志通过Filebeat或Flume写到Kafka，Spark Structured Streaming消费Kafka做实时清洗与窗口聚合，聚合结果一份写明细到HDFS供离线回溯，一份写指标到Redis或ES供在线查询，检测出的异常走告警通道。每一层职责拆开，排障时才不会互相甩锅。
采集层只负责格式化和推送，不要做复杂清洗。清洗规则写死在采集端，后面想加字段要重新发布agent；正确做法是采集层统一输出JSON，字段解析和过滤全部下沉到Spark任务里。消息层Kafka的作用是削峰与缓冲，日志洪峰时Spark消费不过来，Kafka先兜住数据；注意分区数不要无脑调大，Kafka分区数要与Spark并发度匹配，分区太多微批的调度开销反而变大。计算层就是Structured Streaming，负责解析、过滤、窗口聚合、异常判定。存储层明细走HDFS或对象存储，指标走ES和MySQL，告警走钉钉、邮件或webhook。
2.3 集群资源与spark-shell验证的最小配置
一个可参考的下限是3台节点：1台master、2台worker，每台16G内存4核，Kafka和HDFS复用同一批机器。日志量没到单日亿级之前，加机器不如把executor内存和并行度调好。搭建阶段不要直接上spark-submit，先用spark-shell读一段Kafka数据打通链路，再逐步加逻辑。
./bin/spark-shell --master yarn --deploy-mode client \ --driver-memory 2g \ --executor-memory 4g \ --executor-cores 2 \ --num-executors 4 \ --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.2--packages把Kafka数据源依赖拉进来，版本号里的2.12对应Scala编译版本，要和Spark主版本匹配；升级Spark时这个坐标要跟着主版本走。executor-memory 4g按每个executor处理两个分区数据估算，如果日志单条带上堆栈达到几十KB，这里要往上调。spark-shell验证通过再写正式代码，能省一半调试时间。
3. Structured Streaming接Kafka：最小可运行的清洗、窗口聚合与输出代码
3.1 读Kafka并解析日志：bootstrap与schema设置
先把基础数据源跑起来。日志在Kafka里以JSON字符串存放，下面是完整读取代码，整个实时任务的入口基本固定，后续所有逻辑都挂在parsedDF后面。
import org.apache.spark.sql.SparkSession import org.apache.spark.sql.types._ import org.apache.spark.sql.functions._ val spark = SparkSession.builder() .appName("RealtimeLogAnalyzer") .config("spark.sql.shuffle.partitions", "8") .config("spark.streaming.kafka.maxRatePerPartition", "1000") .getOrCreate() val rawDF = spark.readStream .format("kafka") .option("kafka.bootstrap.servers", "kafka01:9092,kafka02:9092") .option("subscribe", "app-log") .option("startingOffsets", "latest") .option("failOnDataLoss", "false") .option("maxOffsetsPerTrigger", "20000") .load() val schema = StructType(Seq( StructField("ts", StringType), StructField("appId", StringType), StructField("level", StringType), StructField("uid", StringType), StructField("path", StringType), StructField("respTime", LongType) )) val parsedDF = rawDF .selectExpr("CAST(value AS STRING) as jsonStr") .select(from_json(col("jsonStr"), schema).as("data")) .select("data.*") .withColumn("eventTime", to_timestamp(col("ts"), "yyyy-MM-dd HH:mm:ss"))readStream.format("kafka")声明这是一个持续运行的流式读，subscribe指定topic，多个topic用逗号分隔。startingOffsets设成latest表示只消费新数据，调试期需要回看历史时改成earliest。from_json把字符串解析成结构化列，schema字段名必须与日志JSON完全一致，不一致的结果是全列为null，而且不报错。to_timestamp的格式串写错会导致整列为空，后面窗口聚合的全是空值，只出脏结果不抛异常，这类问题排查起来很费时间。
failOnDataLoss默认是true，Kafka里日志超过保留期被清理时任务会直接报错退出，生产环境设成false更稳妥；代价是被清理的offset段数据会静默跳过，所以这个参数要配合offset堆积告警一起用。
3.2 事件时间窗口与水位线：参数怎么设
日志分析里最常用的两个粒度是1分钟监控和5分钟趋势，代码用window和withWatermark组合表达。注意这里统计的是事件发生时间，不是Spark收到日志的时间。
val metricsDF = parsedDF .withWatermark("eventTime", "2 minutes") .groupBy( window(col("eventTime"), "1 minute", "1 minute"), col("appId") ) .agg( count("*").as("reqCount"), sum(when(col("respTime") > 1000, 1).otherwise(0)).as("slowCount"), avg("respTime").as("avgRespTime"), approx_count_distinct("uid").as("uv") )withWatermark("eventTime", "2 minutes")声明允许事件时间比处理时间最多晚2分钟，超过这个范围的迟到数据会被丢弃；同时它开启了聚合状态的自动清理，过期窗口状态在水位线推进后自动删除。不设置水位线，聚合状态会一直在内存里膨胀，跑几天后任务从GC异常恶化到OOM。window第一个参数是事件时间列，第二个是窗口长度，第三个是滑动间隔，都设1分钟就表示每1分钟出一个独立桶，滚动窗口不会重复计数，下游逻辑更干净。
approx_count_distinct做UV近似统计，误差在2%以内，相比countDistinct状态量小一个数量级，能明显减轻状态存储压力。生产环境我一般把窗口和滑动间隔设为相同值；滑动窗口曲线更平滑，但同一事件会落入多个窗口，下游异常检测要处理重复计数，容易出事。滚动窗口跑通后再评估要不要换。
3.3 foreachBatch写结果：明细落盘与指标入库
聚合结果有两条去向：明细和指标。明细为了回溯分析写Parquet到HDFS，指标为了在线查询和告警写Redis。用foreachBatch在同一个输出里同时做多件事，这是Structured Streaming里最实用的出口。
val query = metricsDF.writeStream .foreachBatch { (batchDF: DataFrame, batchId: Long) => batchDF.cache() // 明细落HDFS，按天和appId分区 batchDF.write .mode("append") .partitionBy("appId", "day") .parquet("hdfs:///warehouse/log_metrics") // 关键指标写Redis，带TTL一小时 batchDF.foreach { row => val redis = new Jedis("redis-host", 6379) val key = s"metrics:${row.getString(0)}:${row.getLong(1)}" redis.hset(key, "reqCount", row.getLong(2).toString) redis.expire(key, 3600) redis.close() } batchDF.unpersist() } .option("checkpointLocation", "hdfs:///checkpoint/log-analyzer") .queryName("log-metrics") .start()batchDF.cache()避免同一批数据被重复读取两遍，明细写完后再unpersist()释放。Parquet写入用append模式，批与批之间互不覆盖。Redis写入的foreach每行都新建连接，数据量上来会成为瓶颈；生产代码要改成连接池或先collect成列表批量写，这里的写法只做结构说明，扛不住线上流量。
checkpointLocation必须指向HDFS或对象存储这类持久化位置，不能放本地磁盘。任务重启后状态丢失去消费老offset，数据会重复写入下游。另外每次改代码，状态存储的schema尽量不要变，这个坑在第5章详细讲。
4. 异常检测的落地写法：阈值基线、EWMA与联合判定
4.1 周期性3σ基线：读历史、动态更新
异常检测第一步是定义什么叫正常。日志指标是典型的时间序列异常检测场景，中午和凌晨的请求量天差地别，拿全天均值做阈值一定误报。正确做法是取前7天同一分钟的历史数据算均值和标准差，当前值跟同时刻的均值做比较，偏差超过3σ判异常。这套基线逻辑直接放在foreachBatch里做。
def detectBySigma(batchDF: DataFrame): DataFrame = { val today = java.time.LocalDate.now().toString val yesterdayStart = java.time.LocalDateTime.now() .minusDays(1).format(tsFormat) val historyDF = spark.read .parquet("hdfs:///warehouse/log_metrics") .filter(col("day") >= yesterdayStart) .withColumn("minute", date_format(col("windowStart"), "HH:mm")) .groupBy("appId", "minute") .agg( avg("avgRespTime").as("mean"), stddev("avgRespTime").as("std"), count("*").as("sampleCnt") ) batchDF .withColumn("minute", date_format(col("window"), "HH:mm")) .join(historyDF, Seq("appId", "minute"), "left") .withColumn("isAnomaly", when(col("sampleCnt") < 3, lit(false)) .otherwise( abs(col("avgRespTime") - col("mean")) > lit(3.0) * col("std") )) }历史基线从HDFS指标明细读取，用date_format把窗口时间截到分钟维度对齐。sampleCnt < 3时样本太少算出的σ没有意义，直接放行，避免冷启动阶段全误报。σ倍数取3.0，调大误报少漏报多，调小相反；建议先设3.0跑一周，再根据告警记录调整。join用left是因为新上线的模块在昨天没有对应分钟历史，mean会是null，生产代码要用coalesce先兜默认值，否则整列null导致检测静默失效。
这套3σ的局限性是只能捕获幅值突变，缓慢漂移会逐渐被基线“吸收”变得不异常，需要第二种检测来补。
4.2 EWMA残差检测：对均值漂移更敏感
EWMA（指数加权移动平均）在实时检测里非常实用：给近期数据更高权重，让均值跟随正常波动，当前值与EWMA预测值的残差超过阈值就告警。相比3σ，它更适合捕捉缓慢趋势，比如日志量一小时比一小时涨，单点看都算正常，但整体已经脱离历史模式。
class EWMADetector(alpha: Double = 0.3, threshold: Double = 3.0) { private var ewma: Double = Double.NaN private var lastTs: Long = -1L def detect(appId: String, minute: Long, value: Double): Boolean = { if (lastTs == -1L || minute - lastTs > 5) { // 断流超过5分钟，状态重置 ewma = value } else { val residual = math.abs(value - ewma) ewma = alpha * value + (1 - alpha) * ewma if (residual > threshold * estimateStd()) return true } lastTs = minute false } }alpha控制平滑程度，0.3表示新值占30%权重，越高对变化反应越快，对噪声也越敏感；0.1到0.3是常用区间，建议拿历史数据回放定值。minute - lastTs > 5处理长时间无数据的情况：超过5分钟没来数据说明窗口断流，ewma状态已过期，重新初始化，避免用旧状态判断新数据。estimateStd()生产里用EWMA残差的移动标准差代替，训练期先跑一天收集残差分布。
流式任务里这个状态最好托管给StateStore，而不是放在executor对象内部——executor重启后对象状态归零，检测会有一段盲区。用mapGroupsWithState或flatMapGroupsWithState能把状态交给Checkpoint，代价是代码结构复杂一些，但实时检测的连续性有保障。如果第一版先跑通，用对象内状态是可以接受的，但要清楚重启后会有几分钟盲区。
4.3 多维度联合判定与误报收敛
单个指标抖动太常见了，要联合多个指标看。比如只告警响应时间异常，很可能因为一个慢SQL拖慢整个接口误报一晚上；加上错误率和请求量一起判定，误报会明显下降。
val finalAlert = metricsDF .join(sigmaResult, Seq("appId", "window")) .join(ewmaResult, Seq("appId", "window")) .withColumn("alertLevel", when(col("isSigmaAnomaly") && col("isEwmaAnomaly"), lit("P1")) .when(col("isSigmaAnomaly") || col("isEwmaAnomaly"), lit("P2")) .when(col("errorRate") > 0.05 && col("reqCount") > 100, lit("P2")) .otherwise(lit(null)) ) .filter(col("alertLevel").isNotNull)P1代表两种算法同时确认异常，这种告警基本不用复核直接发；P2是单一算法命中，发出来给值班人员参考。errorRate > 0.05 && reqCount > 100是典型联合条件：错误率超5%且请求量超100才有统计意义，请求量只有两条时错误率50%也不该触发。这套规则参数建议做成配置文件，不要硬编码在代码里；调优阶段一天改十几次阈值，每改一次都要编译提交的话，体验非常折磨。
判定完要做告警收敛。窗口是1分钟粒度，一个问题往往连续触发十几条告警；做法是按appId加检测类型做5分钟冷却，冷却时间内同类告警合并成一条，只更新触发次数和最后触发时间。告警通道如果是钉钉或邮件，连续轰炸会导致整个团队把消息设成免打扰——这后果比漏报更严重。
5. 避坑：实时任务崩掉、漏报、结果跳变的常见问题
5.1 事件时间乱序让窗口结果反复跳变
现象：监控图上同一个分钟的请求量先涨到8000，十分钟后掉回6000，再过几分钟又变成6500，像是数据在反复横跳。
原因：日志在应用侧生成本身就有网络延迟和批量上报，Kafka里的到达顺序与事件发生顺序天然不一致。Structured Streaming的窗口聚合基于事件时间，乱序数据到达后会被追加到之前的窗口，已输出过的窗口结果被更新，下游看到的就是跳变。
解决：withWatermark的迟到容忍时间设到窗口长度的2到4倍；同时告警判定延迟执行，不要等窗口关闭瞬间就发，等水位线推进确认窗口结束再做最终判定。
5.2 failOnDataLoss与offset丢失造成的任务自杀
现象：任务跑了两周，某天凌晨突然失败退出，日志报Trying to access offset N but the earliest available offset is M，任务完全停机，Kafka积压越来越多。
原因：Kafka topic日志默认保留7天，如果业务侧长时间没有写入或消费进度落后太多，对应offset已被清理。任务启动发现要消费的offset不存在，默认行为是直接报错，Spark微批任务连续失败几次就整体退出。
解决：failOnDataLoss设成false，让任务启动时从最近可用offset继续；同时topic保留时间要大于任务允许的最大停机时间。注意这个参数不是后悔药，它只是让任务继续跑，被清掉的数据已经丢了，要配合监控Kafka积压量的告警一起用。
5.3 checkpoint状态与代码升级的兼容性
现象：优化了检测逻辑重新提交，任务启动报ClassCastException或StreamMetadata格式不匹配，一直刷屏起不来。
原因：checkpoint里保存了算子状态，包括聚合key类型、schema、自定义类的序列化格式。直接改代码后状态存储的结构对不上，反序列化阶段就崩了。Spark流任务和离线任务不一样，不能随手改完就重启——状态长在checkpoint里。
解决：修改逻辑前先改应用名或换新checkpoint目录做验证，跑通了再切流量；如果变更涉及聚合字段或窗口逻辑，正确流程是停机、改checkpoint目录、消费位点设成latest，宁可丢一小段数据也不能让状态和代码冲突。团队实践上会把checkpoint目录和代码版本绑定挂到CI，提交记录里能看出哪版代码对应哪个目录。
5.4 executor内存不足与告警风暴
现象：任务运行期间告警成片触发，日志显示Container killed by YARN for exceeding memory limits，executor一连串被杀，任务不断重启。
原因：两个独立问题叠加。一是executor内存设置偏小，日志量大时无法容纳处理中的数据和聚合状态，YARN判定超内存杀掉容器；二是检测阈值太敏感，系统抖动时大量指标同时越阈值，告警通道瞬间被淹没。
解决：内存配比上注意driver内存、executor内存和off-heap内存的区别。Spark Streaming场景executor内存4g到8g起步，spark.memory.fraction默认0.6不用动，先调executor数量比调大单个executor内存更有效。告警风暴靠联合判定和冷却时间压制：多算法同时命中才发P1，单算法命中只记录不打扰；同类告警冷却期内合并，低峰期系统自动收敛，第二天上班看汇总。连续三次触发同一个告警才升级人工处理，这条规则能过滤掉大部分网络抖动和发布上线造成的误报。
5.5 解析失败的数据行静默丢弃
现象：告警没触发，但离线对账发现原始日志条数和入库条数差了一截，翻遍任务日志找不到任何异常。
原因：from_json解析失败时，Spark不会报错，而是给整行null。后续聚合里count不会统计这行，数据就悄悄丢了。等业务侧发现问题来查时，Kafka里的原始数据可能已经过期清理了。
解决：解析后加一个校验列，filter(col("data").isNotNull)，同时把parse失败的行单独写入一个“脏数据”topic或表，每天对账用。数据质量是这个系统的地基，地基歪了，上面所有检测算法都是空中楼阁。
6. 验证检测效果与交付：回放实验、告警冷却、文档怎么组织
6.1 回放历史日志验证召回率
异常检测做完，第一件事不是上线，是找一个已发生过的故障时间窗做回放。做法是取故障前7天日志作为基线，把故障当天的日志按时间顺序重放进Kafka，统计三个数字：故障发生到首次告警的间隔、告警里有多少条真正对应故障、故障期间有没有完全漏掉的时段。这三个数字直接决定这套检测能不能交付。
回放时要按真实速度注入，不要全速灌入。全速灌入会让窗口统计节奏与当时完全不同，测出的延迟没有参考价值。用限速把日志按原始时间戳延迟注入，跑完比较检测输出时间与原始故障时间戳的差值，就是这套系统在当前数据量下的实际反应时间，可以作为验收依据。
6.2 冷却时间与告警聚合
每分钟一条告警等于没有告警。我习惯把告警链路拆成两层：检测层只负责判定，通知层负责聚合。检测层命中记录写入告警事件表，通知层每5分钟扫一次，同一个appId加检测类型在冷却窗口内只发一条，附带触发次数和指标变化趋势。这个设计同时解决告警轰炸和错过后可追溯两个问题，值班手机不会被刷屏，事后有完整事件记录。
6.3 代码包与文档怎么组织
交付时源代码按ingestion、processing、detection、output四个模块分清楚，每个模块一个入口类，配置抽成外部文件；保留一个spark-shell调试入口，方便接手的人直接查数据。文档说明按部署手册、参数表、设计说明三个文件组织：部署手册写到从裸机开始跑通全流程的命令级别；参数表里每个参数写默认值、建议值和调整理由；设计说明讲清楚为什么选这套方案而不是别的。参数表里特别留一列注意事项，把上述这些坑的排查思路都放进去，接手的人不用从头踩一遍。
我自己的教训是：这套系统第一次上线时阈值定得太严，线上跑了三天一个告警都没有。第四天凌晨响应时间飙到三倍才弹出来，中间其实经历了近两个小时的缓慢劣化。后来才把EWMA和3σ基线搭配起来——3σ抓幅度突变，EWMA抓缓慢漂移，两种形态才算都覆盖住。做异常检测不要指望一个算法管所有状况，组合策略加留足验证时间，这两件事比调参本身重要。希望帮到你。
本文还有配套的精品资源，点击获取
