尧图网站建设 
简介这份资源是面向计算机、大数据、人工智能等专业学生与技术学习者的分布式实时日志分析与入侵检测系统完整项目包基于Flume采集日志、Spark进行流式处理、Flask搭建可视化与接口层适合用作课程设计、期末大作业或毕业设计的参考方案也可作为学习分布式日志管道与安全检测思路的实战素材。压缩包共108个文件约18.88MB包含Scala源码、Java代码、Python脚本、HTML与JS前端页面、CSS样式、properties与conf配置、sbt构建文件以及日志样本、缓存与编译输出等运行痕迹覆盖采集、计算、展示的完整链路。目前已有234人学习下载。项目代码经过调试下载后可直接运行读者可据此理解Flume与Spark的对接方式、日志字段解析流程、入侵检测规则或模型的组织形式以及Flask端如何呈现分析结果并参考现有目录结构快速定位各模块职责为二次开发与排错提供清晰起点。1. 从一条 SSH 爆破日志说起FlumeSparkFlask 到底在拼什么凌晨两点一台对外暴露的跳板机在 40 分钟内被扫了 1.2 万次 SSH 登录/var/log/secure里密密麻麻的Failed password里混着几条Accepted。运维第二天早上才看到因为日志是事后grep出来的。这个场景就是「基于 FlumeSparkFlask 的分布式实时日志分析与入侵检测系统」要解决的问题让日志在产生的秒级内被采集、被规则和统计模型判定、被可视化告警而不是躺在磁盘里等你去翻。这套组合的分工很清晰。Flume 负责把分散在多台机器上的日志文件、Syslog、应用输出稳定地收上来解决「日志在哪、怎么不丢」Spark 负责在采集流上做窗口聚合、特征提取和异常判定解决「量大、要算得快」Flask 负责把检测结果、告警列表、统计图表暴露成网页解决「人怎么看、怎么查」。三者串起来就是一条从日志产生到告警呈现的实时链路。它适合谁适合手里已经有几台服务器、想给现有业务加一层轻量入侵检测的安全运维和后台开发也适合做大数据课程设计、想找一个能跑通全链路又不过度复杂的实战项目的人。不适合指望它替代商业 SIEM 的场景——它的价值在于可控、可改、能看懂每一行判定逻辑。下面按「先立住原理、再动手复现、最后讲坑」的顺序拆开讲。2. 链路选型与数据流为什么是 Flume 而不是直接写 Kafka2.1 三个组件各自的边界与衔接方式先把数据流画清楚后面所有配置才有依据。典型链路是业务机上的日志文件 → Flume AgentSource 读文件、Channel 缓冲、Sink 输出→ 消息中间层或直接落地 → Spark 消费并计算 → 结果写存储 → Flask 读取展示。Flume 在这里的角色是「采集端」。它的 Source 常见有三类exec适合追一个持续输出的命令spooldir适合监控一个目录里不断新增的文件taildir适合监控多个按行追加的日志文件并记录读取位置。生产里我一般用taildir因为它支持断点续读Agent 重启后不会从头再读一遍也不会漏掉重启期间新增的行。Channel 是 Flume 的缓冲memory快但断电丢数据file慢但能持久化。日志采集这种「丢一条可能就漏一次攻击」的场景建议用filechannel代价是磁盘 IO 和延迟略高。Sink 决定数据往哪走常见是avro发给下一跳 Agent、kafka进消息队列、hdfs离线归档。实时检测要低延迟通常 Sink 到 Kafka再由 Spark Structured Streaming 消费。Spark 的角色是「计算端」。它消费流数据后做两件事一是基于规则的匹配比如某 IP 在 60 秒内失败登录超过 20 次就判定为爆破二是基于统计的异常比如某账号在非活跃时段的登录频率突增。Structured Streaming 用window做滑动窗口聚合天然适合「一段时间内次数」这类判定。Flask 的角色是「展示端」。它不参与计算只从结果存储MySQL、Redis、Elasticsearch 都行里读检测结果和告警渲染成表格和图表。轻量场景用 Flask SQLAlchemy ECharts 就够了不必上重型前端框架。2.2 为什么不用「Flume 直连 Spark」而要多一跳有人会问Flume 的 Sink 能不能直接对接 Spark Streaming技术上可以用avrosink 配合 Spark 的FlumeInputDStream。但这条老路在新版本里维护成本高且 Flume 和 Spark 的版本耦合紧升级一方就容易翻车。更稳的做法是中间加一层 KafkaFlume 只管往 Kafka 写Spark 只管从 Kafka 读两边解耦各自升级互不影响还能利用 Kafka 的分区和副本做削峰与容错。代价是多维护一个 Kafka 集群。如果只是单机实验或课程设计可以退化成 Flume 写本地文件、Spark 读文件流省掉 Kafka但会失去实时性和多消费者能力。选型时按「是否要上生产」这条线切实验环境可以省生产环境别省。2.3 检测逻辑放在 Spark 里而不是 Flume 拦截器里Flume 支持 Interceptor能在采集时做简单过滤和打标签。但把入侵检测规则写进 Interceptor 是常见误用Interceptor 是逐事件同步处理复杂规则会拖慢采集而且它无状态做不了「60 秒内累计次数」这种窗口统计。正确分工是 Interceptor 只做轻量预处理比如补主机名、过滤明显噪声真正的窗口聚合和判定交给 Spark。这样采集端保持轻快计算端承担复杂逻辑。3. 把 Flume 采集端配起来taildir 到 Kafka 的最小可用配置3.1 Agent 配置文件与关键参数下面是一份可直接改用的 Flume Agent 配置采集多个日志文件并写入 Kafka。文件名假设为log-agent.conf。# log-agent.conf a1.sources r1 a1.channels c1 a1.sinks k1 # Source监控多个按行追加的日志文件记录读取位置 a1.sources.r1.type TAILDIR a1.sources.r1.positionFile /data/flume/taildir_position.json a1.sources.r1.filegroups f1 f2 a1.sources.r1.filegroups.f1 /var/log/app/.*\.log a1.sources.r1.filegroups.f2 /var/log/secure # 每行作为一个 event避免多行日志被拆散 a1.sources.r1.interceptors i1 a1.sources.r1.interceptors.i1.type regex_filter a1.sources.r1.interceptors.i1.regex .*Failed password.*|.*Accepted.* a1.sources.r1.interceptors.i1.excludeEvents false # Channel文件通道断电不丢 a1.channels.c1.type file a1.channels.c1.checkpointDir /data/flume/checkpoint a1.channels.c1.dataDirs /data/flume/data a1.channels.c1.capacity 100000 a1.channels.c1.transactionCapacity 1000 # Sink写入 Kafka a1.sinks.k1.type org.apache.flume.sink.kafka.KafkaSink a1.sinks.k1.kafka.bootstrap.servers kafka1:9092,kafka2:9092 a1.sinks.k1.kafka.topic raw-logs a1.sinks.k1.kafka.flumeBatchSize 500 a1.sinks.k1.kafka.producer.acks 1 a1.sources.r1.channels c1 a1.sinks.k1.channel c1逻辑说明TAILDIR通过positionFile记录每个文件的读取偏移重启后从上次位置继续这是它比exec更适合生产日志的关键。filegroups用正则匹配文件新增文件会被自动纳入监控。regex_filter拦截器在这里只保留含Failed password或Accepted的行减少下游压力——注意这是轻量过滤不是检测。参数说明capacity是 Channel 最多缓存多少 eventtransactionCapacity是单次事务最多取多少后者必须小于等于前者否则启动报错。flumeBatchSize控制 Sink 每批写 Kafka 的条数调大吞吐高但延迟增500 是常见折中。acks1表示 leader 写入即确认追求吞吐要更强不丢可设acksall代价是延迟。3.2 启动与验证采集是否通配置写好后用下面的命令启动 Agent并确认数据确实进了 Kafka。# 启动 Flume Agent bin/flume-ng agent \ --conf conf \ --conf-file conf/log-agent.conf \ --name a1 \ -Dflume.root.loggerINFO,console # 另开终端消费 Kafka 验证是否有数据 bin/kafka-console-consumer.sh \ --bootstrap-server kafka1:9092 \ --topic raw-logs \ --from-beginning \ --max-messages 5逻辑说明第一条命令以前台方式启动日志打到控制台方便调试生产里应改成后台并配日志轮转。第二条从 Kafka 读 5 条能读到内容说明 Flume→Kafka 链路通了。如果读不到先看 Flume 控制台有没有SinkException再看 Kafka topic 是否已创建、bootstrap.servers是否可达。参数说明--name a1必须和配置文件里的 Agent 名一致写错会报找不到 source。--from-beginning只在验证时用生产消费端不要随便加否则会重复消费历史数据。3.3 采集端的三个必调项第一是positionFile所在目录要有写权限且别放临时目录机器重启被清掉会导致重复采集。第二是filechannel 的dataDirs建议配多个不同磁盘目录Flume 会轮询写入提升 IO 吞吐。第三是regex_filter的正则别写太复杂逐行匹配的正则在日志量大时是 CPU 热点能前置到应用侧过滤的就别丢给 Flume。4. Spark Structured Streaming 做窗口聚合与入侵判定4.1 从 Kafka 读流并解析日志行Spark 侧第一件事是把 Kafka 里的原始行解析成结构化字段。下面是一段 PySpark 代码读取raw-logs主题并抽出时间、IP、事件类型。from pyspark.sql import SparkSession from pyspark.sql.functions import col, regexp_extract, to_timestamp spark SparkSession.builder \ .appName(LogIntrusionDetect) \ .config(spark.sql.shuffle.partitions, 8) \ .getOrCreate() # 从 Kafka 读取原始流 raw spark.readStream \ .format(kafka) \ .option(kafka.bootstrap.servers, kafka1:9092) \ .option(subscribe, raw-logs) \ .option(startingOffsets, latest) \ .load() # 解析 value 为字符串再抽字段 parsed raw.selectExpr(CAST(value AS STRING) AS line) \ .withColumn(ip, regexp_extract(col(line), rfrom (\d\.\d\.\d\.\d), 1)) \ .withColumn(event, regexp_extract(col(line), r(Failed password|Accepted), 1)) \ .withColumn(ts, to_timestamp(regexp_extract(col(line), r^(\w \d \d:\d:\d), 1), MMM d HH:mm:ss))逻辑说明readStream建立流式读取startingOffsetslatest表示只消费启动后的新数据避免实验时被历史数据淹没。regexp_extract从日志行里抽 IP 和事件类型第三个参数1表示取第一个捕获组。to_timestamp把日志里的时间字符串转成时间戳后续窗口聚合要用它做事件时间。参数说明spark.sql.shuffle.partitions默认 200单机实验设成 8 能减少小文件和无谓调度开销集群上按核数调整一般设成总核数的 2~3 倍。startingOffsets生产里常用earliest配合 checkpoint 做断点续读实验用latest更干净。4.2 滑动窗口统计失败登录次数入侵检测的核心判定是「某 IP 在时间窗口内失败次数超阈值」。用窗口聚合实现。from pyspark.sql.functions import window, count, when # 只保留失败事件按 IP 做 60 秒窗口、30 秒滑动 failed parsed.filter(col(event) Failed password) \ .withWatermark(ts, 2 minutes) windowed failed.groupBy( window(col(ts), 60 seconds, 30 seconds), col(ip) ).agg(count(*).alias(fail_cnt)) # 超过 20 次判定为疑似爆破 alerts windowed.withColumn( level, when(col(fail_cnt) 20, HIGH).otherwise(LOW) ).filter(col(level) HIGH)逻辑说明withWatermark定义事件时间的水位线允许迟到 2 分钟的数据仍被计入窗口超过则丢弃防止状态无限增长。window(ts, 60 seconds, 30 seconds)表示窗口长 60 秒、每 30 秒滑动一次即每 30 秒输出一次「过去 60 秒」的统计。count(*)统计窗口内该 IP 的失败次数超过 20 次打 HIGH 标签。参数说明窗口长度和阈值是最需要按业务调的两个数。窗口太短会漏掉慢速爆破太长会误报正常用户的多次输错。阈值 20 是经验起点实际要结合你们业务的正常失败率——如果正常时段每分钟就有几十次失败阈值要相应抬高或者改成「相对基线的突增」而非绝对次数。4.3 结果写出与 checkpoint 配置检测结果要落到存储供 Flask 读取同时 checkpoint 保证重启不丢状态。# 告警写 MySQL通过 foreachBatch 自定义写入 def write_alerts(df, epoch_id): df.select(ip, fail_cnt, level, window) \ .write \ .format(jdbc) \ .option(url, jdbc:mysql://mysql:3306/ids) \ .option(dbtable, alerts) \ .option(user, ids) \ .option(password, ******) \ .mode(append) \ .save() query alerts.writeStream \ .foreachBatch(write_alerts) \ .outputMode(update) \ .option(checkpointLocation, /data/spark/checkpoint/ids) \ .trigger(processingTime30 seconds) \ .start() query.awaitTermination()逻辑说明foreachBatch让每个微批的结果走自定义写入逻辑这里写 MySQL。outputMode(update)只输出有更新的窗口结果适合告警场景。checkpointLocation保存流处理进度和窗口状态重启后从断点继续这是生产必备。参数说明trigger(processingTime30 seconds)表示每 30 秒触发一个微批和窗口滑动步长对齐比较自然。checkpoint 目录要放在可靠存储上且同一查询的 checkpoint 不能被两个作业共用否则状态错乱。5. Flask 展示层与告警查询接口5.1 用 SQLAlchemy 读告警并渲染Flask 侧不参与计算只查询和展示。下面是一个最小可用的告警列表接口和页面渲染。from flask import Flask, render_template from flask_sqlalchemy import SQLAlchemy app Flask(__name__) app.config[SQLALCHEMY_DATABASE_URI] mysqlpymysql://ids:******mysql:3306/ids db SQLAlchemy(app) class Alert(db.Model): __tablename__ alerts id db.Column(db.Integer, primary_keyTrue) ip db.Column(db.String(64)) fail_cnt db.Column(db.Integer) level db.Column(db.String(16)) app.route(/alerts) def alert_list(): # 按失败次数倒序取最近 100 条 rows Alert.query.order_by(Alert.fail_cnt.desc()).limit(100).all() return render_template(alerts.html, alertsrows)逻辑说明Alert模型映射 Spark 写入的alerts表字段名要和写入时一致。alert_list查询最近 100 条按失败次数排序的告警交给模板渲染。模板里用循环把ip、fail_cnt、level显示成表格即可前端图表可另接 ECharts 读同一份数据。参数说明数据库连接串里的账号只给SELECT权限即可展示层不需要写权限降低被利用风险。limit(100)是防止一次拉全表拖垮页面实际可加分页。5.2 给告警加一个按 IP 聚合的统计接口运维更关心「哪个 IP 最活跃」加一个聚合接口。from sqlalchemy import func app.route(/stats/top_ips) def top_ips(): rows db.session.query( Alert.ip, func.sum(Alert.fail_cnt).label(total) ).group_by(Alert.ip) \ .order_by(func.sum(Alert.fail_cnt).desc()) \ .limit(10).all() return {data: [{ip: r.ip, total: int(r.total)} for r in rows]}逻辑说明按 IP 分组求失败次数总和取前 10返回 JSON 供前端图表消费。func.sum是 SQL 聚合计算压力在数据库侧Flask 只做转发。参数说明如果告警表数据量大这个查询会慢应在ip和写入时间上建索引或让 Spark 侧预先聚合好「每 IP 累计次数」写入另一张表Flask 直接读结果。5.3 展示层的部署注意Flask 自带的开发服务器不能上生产用gunicorn起多 worker前面挂 Nginx 做静态资源和反向代理。数据库连接用连接池避免每个请求新建连接。页面上的告警时间要统一时区Spark 写入的时间戳和 Flask 展示的时区不一致是常见「时间对不上」问题建议全链路统一用 UTC 存储、展示时再转本地。6. 避坑与排查这套链路最容易翻车的五个地方6.1 现象Flume 重启后日志重复采集原因positionFile所在目录被清理或配置里没写positionFileAgent 每次启动都从头读。解决把positionFile放到持久化目录并纳入备份确认进程对该文件有读写权限不要用/tmp存放。6.2 现象Spark 作业跑一段时间后内存暴涨直至 OOM原因窗口聚合的状态没有水位线约束迟到数据不断累积或spark.sql.shuffle.partitions过大导致大量小任务。解决给事件时间加withWatermark并设合理迟到容忍把 shuffle 分区数按数据量调小必要时用state相关的超时配置清理长期不更新的 key。6.3 现象Kafka 消费延迟越来越高告警滞后几分钟原因Spark 微批处理时间超过触发间隔积压越来越多或下游 MySQL 写入慢成为瓶颈。解决先看 Spark UI 里每批的处理时长若写入慢就改批量写入或换更快的存储若计算慢就减少窗口数量、优化正则。trigger间隔不要设得比单批处理时间还短否则永远追不上。6.4 现象Flask 页面打开报数据库连接超时原因连接池耗尽或 MySQL 最大连接数被打满也可能是 Spark 写入和 Flask 读取争抢连接。解决给 Flask 配连接池并设上限给 Spark 写入单独账号检查 MySQLmax_connections必要时调大并排查是否有连接泄漏。6.5 现象告警里大量正常登录被误判为爆破原因阈值一刀切没考虑业务正常失败率或窗口太短把用户连续输错密码当成攻击。解决先统计正常时段的失败分布把阈值设在 P99 之上对同一账号的成功登录做白名单抑制把「失败后紧跟成功」的模式单独标记这类往往是用户自己输错而非攻击。7. 让检测更准从固定阈值到基线偏离固定阈值能跑通但上线后你会发现它要么太松要么太紧。更实用的进阶做法是给每个 IP 或账号建一条「正常行为基线」检测偏离而不是检测绝对值。具体做法用 Spark 按小时统计每个源 IP 的历史失败次数均值与标准差实时窗口的失败次数超过「均值 3 倍标准差」才告警。这样业务高峰期阈值自动抬高低谷期自动收紧误报会明显下降。实现上可以起一个批处理作业每天更新基线表流作业在判定时关联这张表。下面是一个简化的基线计算片段。# 批处理按 IP 计算历史失败次数的均值与标准差 from pyspark.sql.functions import avg, stddev baseline spark.read.table(alerts_history) \ .groupBy(ip) \ .agg(avg(fail_cnt).alias(avg_cnt), stddev(fail_cnt).alias(std_cnt)) baseline.write.mode(overwrite).saveAsTable(ip_baseline)流作业里把实时窗口结果和ip_baseline做 join判定条件改成fail_cnt avg_cnt 3 * std_cnt。参数上3 倍标准差是统计上的常用边界实际可按误报容忍度调到 2.5 或 4。要注意冷启动问题新 IP 没有历史基线先按固定阈值兜底积累够样本后再切基线判定。另一个技巧是给告警加「抑制窗口」同一 IP 在 10 分钟内只报一次避免爆破期间刷屏。这可以在 Flask 展示层做去重也可以在 Spark 侧用dropDuplicates配合水位线实现。我自己的习惯是先在流里做粗判把候选告警写出来再在展示层做抑制和分级这样调整策略不用重启流作业。踩过的坑是抑制逻辑写在流里改一次规则就要重启状态还得重建后来统一挪到展示层灵活很多。希望帮到你。本文还有配套的精品资源点击获取 基于Flume+Spark+Flask的分布式实时日志分析与入侵检测系统实战
简介这份资源是面向计算机、大数据、人工智能等专业学生与技术学习者的分布式实时日志分析与入侵检测系统完整项目包基于Flume采集日志、Spark进行流式处理、Flask搭建可视化与接口层适合用作课程设计、期末大作业或毕业设计的参考方案也可作为学习分布式日志管道与安全检测思路的实战素材。压缩包共108个文件约18.88MB包含Scala源码、Java代码、Python脚本、HTML与JS前端页面、CSS样式、properties与conf配置、sbt构建文件以及日志样本、缓存与编译输出等运行痕迹覆盖采集、计算、展示的完整链路。目前已有234人学习下载。项目代码经过调试下载后可直接运行读者可据此理解Flume与Spark的对接方式、日志字段解析流程、入侵检测规则或模型的组织形式以及Flask端如何呈现分析结果并参考现有目录结构快速定位各模块职责为二次开发与排错提供清晰起点。1. 从一条 SSH 爆破日志说起FlumeSparkFlask 到底在拼什么凌晨两点一台对外暴露的跳板机在 40 分钟内被扫了 1.2 万次 SSH 登录/var/log/secure里密密麻麻的Failed password里混着几条Accepted。运维第二天早上才看到因为日志是事后grep出来的。这个场景就是「基于 FlumeSparkFlask 的分布式实时日志分析与入侵检测系统」要解决的问题让日志在产生的秒级内被采集、被规则和统计模型判定、被可视化告警而不是躺在磁盘里等你去翻。这套组合的分工很清晰。Flume 负责把分散在多台机器上的日志文件、Syslog、应用输出稳定地收上来解决「日志在哪、怎么不丢」Spark 负责在采集流上做窗口聚合、特征提取和异常判定解决「量大、要算得快」Flask 负责把检测结果、告警列表、统计图表暴露成网页解决「人怎么看、怎么查」。三者串起来就是一条从日志产生到告警呈现的实时链路。它适合谁适合手里已经有几台服务器、想给现有业务加一层轻量入侵检测的安全运维和后台开发也适合做大数据课程设计、想找一个能跑通全链路又不过度复杂的实战项目的人。不适合指望它替代商业 SIEM 的场景——它的价值在于可控、可改、能看懂每一行判定逻辑。下面按「先立住原理、再动手复现、最后讲坑」的顺序拆开讲。2. 链路选型与数据流为什么是 Flume 而不是直接写 Kafka2.1 三个组件各自的边界与衔接方式先把数据流画清楚后面所有配置才有依据。典型链路是业务机上的日志文件 → Flume AgentSource 读文件、Channel 缓冲、Sink 输出→ 消息中间层或直接落地 → Spark 消费并计算 → 结果写存储 → Flask 读取展示。Flume 在这里的角色是「采集端」。它的 Source 常见有三类exec适合追一个持续输出的命令spooldir适合监控一个目录里不断新增的文件taildir适合监控多个按行追加的日志文件并记录读取位置。生产里我一般用taildir因为它支持断点续读Agent 重启后不会从头再读一遍也不会漏掉重启期间新增的行。Channel 是 Flume 的缓冲memory快但断电丢数据file慢但能持久化。日志采集这种「丢一条可能就漏一次攻击」的场景建议用filechannel代价是磁盘 IO 和延迟略高。Sink 决定数据往哪走常见是avro发给下一跳 Agent、kafka进消息队列、hdfs离线归档。实时检测要低延迟通常 Sink 到 Kafka再由 Spark Structured Streaming 消费。Spark 的角色是「计算端」。它消费流数据后做两件事一是基于规则的匹配比如某 IP 在 60 秒内失败登录超过 20 次就判定为爆破二是基于统计的异常比如某账号在非活跃时段的登录频率突增。Structured Streaming 用window做滑动窗口聚合天然适合「一段时间内次数」这类判定。Flask 的角色是「展示端」。它不参与计算只从结果存储MySQL、Redis、Elasticsearch 都行里读检测结果和告警渲染成表格和图表。轻量场景用 Flask SQLAlchemy ECharts 就够了不必上重型前端框架。2.2 为什么不用「Flume 直连 Spark」而要多一跳有人会问Flume 的 Sink 能不能直接对接 Spark Streaming技术上可以用avrosink 配合 Spark 的FlumeInputDStream。但这条老路在新版本里维护成本高且 Flume 和 Spark 的版本耦合紧升级一方就容易翻车。更稳的做法是中间加一层 KafkaFlume 只管往 Kafka 写Spark 只管从 Kafka 读两边解耦各自升级互不影响还能利用 Kafka 的分区和副本做削峰与容错。代价是多维护一个 Kafka 集群。如果只是单机实验或课程设计可以退化成 Flume 写本地文件、Spark 读文件流省掉 Kafka但会失去实时性和多消费者能力。选型时按「是否要上生产」这条线切实验环境可以省生产环境别省。2.3 检测逻辑放在 Spark 里而不是 Flume 拦截器里Flume 支持 Interceptor能在采集时做简单过滤和打标签。但把入侵检测规则写进 Interceptor 是常见误用Interceptor 是逐事件同步处理复杂规则会拖慢采集而且它无状态做不了「60 秒内累计次数」这种窗口统计。正确分工是 Interceptor 只做轻量预处理比如补主机名、过滤明显噪声真正的窗口聚合和判定交给 Spark。这样采集端保持轻快计算端承担复杂逻辑。3. 把 Flume 采集端配起来taildir 到 Kafka 的最小可用配置3.1 Agent 配置文件与关键参数下面是一份可直接改用的 Flume Agent 配置采集多个日志文件并写入 Kafka。文件名假设为log-agent.conf。# log-agent.conf a1.sources r1 a1.channels c1 a1.sinks k1 # Source监控多个按行追加的日志文件记录读取位置 a1.sources.r1.type TAILDIR a1.sources.r1.positionFile /data/flume/taildir_position.json a1.sources.r1.filegroups f1 f2 a1.sources.r1.filegroups.f1 /var/log/app/.*\.log a1.sources.r1.filegroups.f2 /var/log/secure # 每行作为一个 event避免多行日志被拆散 a1.sources.r1.interceptors i1 a1.sources.r1.interceptors.i1.type regex_filter a1.sources.r1.interceptors.i1.regex .*Failed password.*|.*Accepted.* a1.sources.r1.interceptors.i1.excludeEvents false # Channel文件通道断电不丢 a1.channels.c1.type file a1.channels.c1.checkpointDir /data/flume/checkpoint a1.channels.c1.dataDirs /data/flume/data a1.channels.c1.capacity 100000 a1.channels.c1.transactionCapacity 1000 # Sink写入 Kafka a1.sinks.k1.type org.apache.flume.sink.kafka.KafkaSink a1.sinks.k1.kafka.bootstrap.servers kafka1:9092,kafka2:9092 a1.sinks.k1.kafka.topic raw-logs a1.sinks.k1.kafka.flumeBatchSize 500 a1.sinks.k1.kafka.producer.acks 1 a1.sources.r1.channels c1 a1.sinks.k1.channel c1逻辑说明TAILDIR通过positionFile记录每个文件的读取偏移重启后从上次位置继续这是它比exec更适合生产日志的关键。filegroups用正则匹配文件新增文件会被自动纳入监控。regex_filter拦截器在这里只保留含Failed password或Accepted的行减少下游压力——注意这是轻量过滤不是检测。参数说明capacity是 Channel 最多缓存多少 eventtransactionCapacity是单次事务最多取多少后者必须小于等于前者否则启动报错。flumeBatchSize控制 Sink 每批写 Kafka 的条数调大吞吐高但延迟增500 是常见折中。acks1表示 leader 写入即确认追求吞吐要更强不丢可设acksall代价是延迟。3.2 启动与验证采集是否通配置写好后用下面的命令启动 Agent并确认数据确实进了 Kafka。# 启动 Flume Agent bin/flume-ng agent \ --conf conf \ --conf-file conf/log-agent.conf \ --name a1 \ -Dflume.root.loggerINFO,console # 另开终端消费 Kafka 验证是否有数据 bin/kafka-console-consumer.sh \ --bootstrap-server kafka1:9092 \ --topic raw-logs \ --from-beginning \ --max-messages 5逻辑说明第一条命令以前台方式启动日志打到控制台方便调试生产里应改成后台并配日志轮转。第二条从 Kafka 读 5 条能读到内容说明 Flume→Kafka 链路通了。如果读不到先看 Flume 控制台有没有SinkException再看 Kafka topic 是否已创建、bootstrap.servers是否可达。参数说明--name a1必须和配置文件里的 Agent 名一致写错会报找不到 source。--from-beginning只在验证时用生产消费端不要随便加否则会重复消费历史数据。3.3 采集端的三个必调项第一是positionFile所在目录要有写权限且别放临时目录机器重启被清掉会导致重复采集。第二是filechannel 的dataDirs建议配多个不同磁盘目录Flume 会轮询写入提升 IO 吞吐。第三是regex_filter的正则别写太复杂逐行匹配的正则在日志量大时是 CPU 热点能前置到应用侧过滤的就别丢给 Flume。4. Spark Structured Streaming 做窗口聚合与入侵判定4.1 从 Kafka 读流并解析日志行Spark 侧第一件事是把 Kafka 里的原始行解析成结构化字段。下面是一段 PySpark 代码读取raw-logs主题并抽出时间、IP、事件类型。from pyspark.sql import SparkSession from pyspark.sql.functions import col, regexp_extract, to_timestamp spark SparkSession.builder \ .appName(LogIntrusionDetect) \ .config(spark.sql.shuffle.partitions, 8) \ .getOrCreate() # 从 Kafka 读取原始流 raw spark.readStream \ .format(kafka) \ .option(kafka.bootstrap.servers, kafka1:9092) \ .option(subscribe, raw-logs) \ .option(startingOffsets, latest) \ .load() # 解析 value 为字符串再抽字段 parsed raw.selectExpr(CAST(value AS STRING) AS line) \ .withColumn(ip, regexp_extract(col(line), rfrom (\d\.\d\.\d\.\d), 1)) \ .withColumn(event, regexp_extract(col(line), r(Failed password|Accepted), 1)) \ .withColumn(ts, to_timestamp(regexp_extract(col(line), r^(\w \d \d:\d:\d), 1), MMM d HH:mm:ss))逻辑说明readStream建立流式读取startingOffsetslatest表示只消费启动后的新数据避免实验时被历史数据淹没。regexp_extract从日志行里抽 IP 和事件类型第三个参数1表示取第一个捕获组。to_timestamp把日志里的时间字符串转成时间戳后续窗口聚合要用它做事件时间。参数说明spark.sql.shuffle.partitions默认 200单机实验设成 8 能减少小文件和无谓调度开销集群上按核数调整一般设成总核数的 2~3 倍。startingOffsets生产里常用earliest配合 checkpoint 做断点续读实验用latest更干净。4.2 滑动窗口统计失败登录次数入侵检测的核心判定是「某 IP 在时间窗口内失败次数超阈值」。用窗口聚合实现。from pyspark.sql.functions import window, count, when # 只保留失败事件按 IP 做 60 秒窗口、30 秒滑动 failed parsed.filter(col(event) Failed password) \ .withWatermark(ts, 2 minutes) windowed failed.groupBy( window(col(ts), 60 seconds, 30 seconds), col(ip) ).agg(count(*).alias(fail_cnt)) # 超过 20 次判定为疑似爆破 alerts windowed.withColumn( level, when(col(fail_cnt) 20, HIGH).otherwise(LOW) ).filter(col(level) HIGH)逻辑说明withWatermark定义事件时间的水位线允许迟到 2 分钟的数据仍被计入窗口超过则丢弃防止状态无限增长。window(ts, 60 seconds, 30 seconds)表示窗口长 60 秒、每 30 秒滑动一次即每 30 秒输出一次「过去 60 秒」的统计。count(*)统计窗口内该 IP 的失败次数超过 20 次打 HIGH 标签。参数说明窗口长度和阈值是最需要按业务调的两个数。窗口太短会漏掉慢速爆破太长会误报正常用户的多次输错。阈值 20 是经验起点实际要结合你们业务的正常失败率——如果正常时段每分钟就有几十次失败阈值要相应抬高或者改成「相对基线的突增」而非绝对次数。4.3 结果写出与 checkpoint 配置检测结果要落到存储供 Flask 读取同时 checkpoint 保证重启不丢状态。# 告警写 MySQL通过 foreachBatch 自定义写入 def write_alerts(df, epoch_id): df.select(ip, fail_cnt, level, window) \ .write \ .format(jdbc) \ .option(url, jdbc:mysql://mysql:3306/ids) \ .option(dbtable, alerts) \ .option(user, ids) \ .option(password, ******) \ .mode(append) \ .save() query alerts.writeStream \ .foreachBatch(write_alerts) \ .outputMode(update) \ .option(checkpointLocation, /data/spark/checkpoint/ids) \ .trigger(processingTime30 seconds) \ .start() query.awaitTermination()逻辑说明foreachBatch让每个微批的结果走自定义写入逻辑这里写 MySQL。outputMode(update)只输出有更新的窗口结果适合告警场景。checkpointLocation保存流处理进度和窗口状态重启后从断点继续这是生产必备。参数说明trigger(processingTime30 seconds)表示每 30 秒触发一个微批和窗口滑动步长对齐比较自然。checkpoint 目录要放在可靠存储上且同一查询的 checkpoint 不能被两个作业共用否则状态错乱。5. Flask 展示层与告警查询接口5.1 用 SQLAlchemy 读告警并渲染Flask 侧不参与计算只查询和展示。下面是一个最小可用的告警列表接口和页面渲染。from flask import Flask, render_template from flask_sqlalchemy import SQLAlchemy app Flask(__name__) app.config[SQLALCHEMY_DATABASE_URI] mysqlpymysql://ids:******mysql:3306/ids db SQLAlchemy(app) class Alert(db.Model): __tablename__ alerts id db.Column(db.Integer, primary_keyTrue) ip db.Column(db.String(64)) fail_cnt db.Column(db.Integer) level db.Column(db.String(16)) app.route(/alerts) def alert_list(): # 按失败次数倒序取最近 100 条 rows Alert.query.order_by(Alert.fail_cnt.desc()).limit(100).all() return render_template(alerts.html, alertsrows)逻辑说明Alert模型映射 Spark 写入的alerts表字段名要和写入时一致。alert_list查询最近 100 条按失败次数排序的告警交给模板渲染。模板里用循环把ip、fail_cnt、level显示成表格即可前端图表可另接 ECharts 读同一份数据。参数说明数据库连接串里的账号只给SELECT权限即可展示层不需要写权限降低被利用风险。limit(100)是防止一次拉全表拖垮页面实际可加分页。5.2 给告警加一个按 IP 聚合的统计接口运维更关心「哪个 IP 最活跃」加一个聚合接口。from sqlalchemy import func app.route(/stats/top_ips) def top_ips(): rows db.session.query( Alert.ip, func.sum(Alert.fail_cnt).label(total) ).group_by(Alert.ip) \ .order_by(func.sum(Alert.fail_cnt).desc()) \ .limit(10).all() return {data: [{ip: r.ip, total: int(r.total)} for r in rows]}逻辑说明按 IP 分组求失败次数总和取前 10返回 JSON 供前端图表消费。func.sum是 SQL 聚合计算压力在数据库侧Flask 只做转发。参数说明如果告警表数据量大这个查询会慢应在ip和写入时间上建索引或让 Spark 侧预先聚合好「每 IP 累计次数」写入另一张表Flask 直接读结果。5.3 展示层的部署注意Flask 自带的开发服务器不能上生产用gunicorn起多 worker前面挂 Nginx 做静态资源和反向代理。数据库连接用连接池避免每个请求新建连接。页面上的告警时间要统一时区Spark 写入的时间戳和 Flask 展示的时区不一致是常见「时间对不上」问题建议全链路统一用 UTC 存储、展示时再转本地。6. 避坑与排查这套链路最容易翻车的五个地方6.1 现象Flume 重启后日志重复采集原因positionFile所在目录被清理或配置里没写positionFileAgent 每次启动都从头读。解决把positionFile放到持久化目录并纳入备份确认进程对该文件有读写权限不要用/tmp存放。6.2 现象Spark 作业跑一段时间后内存暴涨直至 OOM原因窗口聚合的状态没有水位线约束迟到数据不断累积或spark.sql.shuffle.partitions过大导致大量小任务。解决给事件时间加withWatermark并设合理迟到容忍把 shuffle 分区数按数据量调小必要时用state相关的超时配置清理长期不更新的 key。6.3 现象Kafka 消费延迟越来越高告警滞后几分钟原因Spark 微批处理时间超过触发间隔积压越来越多或下游 MySQL 写入慢成为瓶颈。解决先看 Spark UI 里每批的处理时长若写入慢就改批量写入或换更快的存储若计算慢就减少窗口数量、优化正则。trigger间隔不要设得比单批处理时间还短否则永远追不上。6.4 现象Flask 页面打开报数据库连接超时原因连接池耗尽或 MySQL 最大连接数被打满也可能是 Spark 写入和 Flask 读取争抢连接。解决给 Flask 配连接池并设上限给 Spark 写入单独账号检查 MySQLmax_connections必要时调大并排查是否有连接泄漏。6.5 现象告警里大量正常登录被误判为爆破原因阈值一刀切没考虑业务正常失败率或窗口太短把用户连续输错密码当成攻击。解决先统计正常时段的失败分布把阈值设在 P99 之上对同一账号的成功登录做白名单抑制把「失败后紧跟成功」的模式单独标记这类往往是用户自己输错而非攻击。7. 让检测更准从固定阈值到基线偏离固定阈值能跑通但上线后你会发现它要么太松要么太紧。更实用的进阶做法是给每个 IP 或账号建一条「正常行为基线」检测偏离而不是检测绝对值。具体做法用 Spark 按小时统计每个源 IP 的历史失败次数均值与标准差实时窗口的失败次数超过「均值 3 倍标准差」才告警。这样业务高峰期阈值自动抬高低谷期自动收紧误报会明显下降。实现上可以起一个批处理作业每天更新基线表流作业在判定时关联这张表。下面是一个简化的基线计算片段。# 批处理按 IP 计算历史失败次数的均值与标准差 from pyspark.sql.functions import avg, stddev baseline spark.read.table(alerts_history) \ .groupBy(ip) \ .agg(avg(fail_cnt).alias(avg_cnt), stddev(fail_cnt).alias(std_cnt)) baseline.write.mode(overwrite).saveAsTable(ip_baseline)流作业里把实时窗口结果和ip_baseline做 join判定条件改成fail_cnt avg_cnt 3 * std_cnt。参数上3 倍标准差是统计上的常用边界实际可按误报容忍度调到 2.5 或 4。要注意冷启动问题新 IP 没有历史基线先按固定阈值兜底积累够样本后再切基线判定。另一个技巧是给告警加「抑制窗口」同一 IP 在 10 分钟内只报一次避免爆破期间刷屏。这可以在 Flask 展示层做去重也可以在 Spark 侧用dropDuplicates配合水位线实现。我自己的习惯是先在流里做粗判把候选告警写出来再在展示层做抑制和分级这样调整策略不用重启流作业。踩过的坑是抑制逻辑写在流里改一次规则就要重启状态还得重建后来统一挪到展示层灵活很多。希望帮到你。本文还有配套的精品资源点击获取