簡(jiǎn)介這份資源是面向計(jì)算機(jī)、大數(shù)據(jù)、人工智能等專業(yè)學(xué)生與技術(shù)學(xué)習(xí)者的分布式實(shí)時(shí)日志分析與入侵檢測(cè)系統(tǒng)完整項(xiàng)目包基于Flume采集日志、Spark進(jìn)行流式處理、Flask搭建可視化與接口層適合用作課程設(shè)計(jì)、期末大作業(yè)或畢業(yè)設(shè)計(jì)的參考方案也可作為學(xué)習(xí)分布式日志管道與安全檢測(cè)思路的實(shí)戰(zhàn)素材。壓縮包共108個(gè)文件約18.88MB包含Scala與Java源碼、sbt構(gòu)建配置、properties與conf配置文件、HTML/CSS/JS前端頁(yè)面、Python腳本以及日志與數(shù)據(jù)樣本等覆蓋采集、計(jì)算、展示各環(huán)節(jié)目錄結(jié)構(gòu)便于按模塊查閱。目前已有234人學(xué)習(xí)下載。項(xiàng)目代碼經(jīng)過(guò)調(diào)試下載后可直接運(yùn)行讀者可據(jù)此理解Flume到Spark再到Flask的完整數(shù)據(jù)鏈路掌握日志解析、實(shí)時(shí)統(tǒng)計(jì)與入侵行為識(shí)別的實(shí)現(xiàn)方式并參考其中的配置與排錯(cuò)思路快速搭建自己的實(shí)驗(yàn)環(huán)境。1. 從一堆.cache文件說(shuō)起這套 FlumeSparkFlask 日志入侵檢測(cè)系統(tǒng)到底能跑出什么如果你手頭正好有一個(gè)「基于 FlumeSparkFlask 的分布式實(shí)時(shí)日志分析與入侵檢測(cè)系統(tǒng)」的壓縮包解壓后第一眼看到的很可能不是熟悉的.py或.java而是一串像access_log、$3d85af9b26c1a259b49e.cache、$da50ce791668c9ed0f15$.class這樣的文件。別慌這不是打包出錯(cuò)而是 Spark 在本地或集群模式下運(yùn)行時(shí)留下的中間產(chǎn)物——.cache是 RDD 或 DataFrame 被persist()后落盤的塊文件$.class則是 Scala 編譯出的匿名類。能出現(xiàn)這些文件說(shuō)明這套代碼至少被真實(shí)提交運(yùn)行過(guò)不是純靜態(tài)的「骨架工程」。這套資源解決的是一個(gè)很具體的問(wèn)題把分散在多臺(tái)機(jī)器上的訪問(wèn)日志通過(guò) Flume 采集匯聚交給 Spark 做實(shí)時(shí)解析和規(guī)則匹配識(shí)別出暴力破解、異常高頻訪問(wèn)、可疑路徑掃描等入侵特征最后用 Flask 提供一個(gè)能看圖表和告警的 Web 界面。它適合正在做課程設(shè)計(jì)、期末大作業(yè)或畢設(shè)的計(jì)算機(jī)、大數(shù)據(jù)、人工智能方向的學(xué)生也適合想跑通「采集→計(jì)算→展示」完整鏈路的技術(shù)學(xué)習(xí)者。前提是你得有一點(diǎn) Linux、Java 和 Python 基礎(chǔ)否則連 Flume 的配置文件都改不動(dòng)。2. 拆開(kāi)壓縮包先看什么Flume、Spark、Flask 三層各自的入口與配置2.1 目錄結(jié)構(gòu)與三個(gè)核心入口拿到壓縮包后不要急著pip install先把目錄樹(shù)看清楚。這類項(xiàng)目通常按技術(shù)棧分層常見(jiàn)結(jié)構(gòu)是flume-conf/、spark-job/、flask-web/三個(gè)主目錄外加一個(gè)logs/放模擬日志、一個(gè)sql/放建表語(yǔ)句。你要找的第一個(gè)文件是 Flume 的.conf配置第二個(gè)是 Spark 的提交腳本或main函數(shù)第三個(gè)是 Flask 的app.py或run.py。先確認(rèn)三件事Flume 的 source 類型是exec還是taildirSpark 的入口是SparkSession還是老的SparkContextFlask 是直接讀 Spark 寫出的結(jié)果表還是通過(guò) API 再查一次。這三個(gè)選擇決定了你后面要不要裝 Kafka、要不要配 Hive、要不要起 Redis。很多同學(xué)跑不起來(lái)不是代碼錯(cuò)而是沒(méi)意識(shí)到這套工程默認(rèn)依賴了外部存儲(chǔ)。# 先看目錄層級(jí)確認(rèn)三個(gè)入口文件的位置 find . -maxdepth 3 -type f \( -name *.conf -o -name *.py -o -name *.scala -o -name *.sql \) | sort # 看 Flume 配置里 source、channel、sink 分別是什么 grep -E a1\.(sources|channels|sinks) flume-conf/*.conf # 看 Spark 作業(yè)的提交方式是 spark-submit 還是 python 直接跑 head -50 spark-job/*.py 2/dev/null || head -50 spark-job/*.scala 2/dev/null上面三條命令的作用分別是定位所有可能的入口文件、提取 Flume 的組件聲明、判斷 Spark 作業(yè)的語(yǔ)言和提交方式。參數(shù)上重點(diǎn)看a1.sources.r1.type如果是TAILDIR就支持?jǐn)帱c(diǎn)續(xù)傳如果是EXEC則每次重啟會(huì)從頭讀生產(chǎn)環(huán)境一般選前者。a1.sinks.k1.type如果是logger說(shuō)明只是調(diào)試用真正落地通常改成hdfs或kafka。2.2 Flume 采集配置source、channel、sink 怎么改才不丟數(shù)據(jù)Flume 這一層最容易翻車的地方是 channel 容量和 batchSize 不匹配。默認(rèn)capacity1000、transactionCapacity100如果日志突發(fā)流量大source 寫入速度超過(guò) sink 消費(fèi)速度channel 滿了就會(huì)拋ChannelException日志直接丟。常見(jiàn)做法是把capacity調(diào)到 10000 以上transactionCapacity調(diào)到 1000同時(shí)把 sink 的batchSize設(shè)成和transactionCapacity一致。# flume-conf/access-log.conf 關(guān)鍵參數(shù) a1.sources.r1.type TAILDIR a1.sources.r1.positionFile /tmp/flume_taildir_position.json a1.sources.r1.filegroups.f1 /home/logs/access.log.* a1.sources.r1.batchSize 1000 a1.channels.c1.type memory a1.channels.c1.capacity 20000 a1.channels.c1.transactionCapacity 2000 a1.sinks.k1.type org.apache.flume.sink.kafka.KafkaSink a1.sinks.k1.kafka.topic access_log_topic a1.sinks.k1.kafka.bootstrap.servers localhost:9092 a1.sinks.k1.batchSize 2000這段配置的邏輯是TAILDIR按文件組監(jiān)控日志positionFile記錄讀取偏移量重啟后不會(huì)重復(fù)消費(fèi)。channel 容量給到 20000事務(wù)容量 2000sink 的 batchSize 也設(shè) 2000三者形成背壓緩沖。如果不想引入 Kafka把 sink 改成hdfs或logger也能跑但實(shí)時(shí)性會(huì)打折扣。注意positionFile的路徑要有寫權(quán)限否則 Flume 啟動(dòng)時(shí)會(huì)靜默失敗日志里只報(bào)一行Permission denied。2.3 Spark 實(shí)時(shí)解析從日志行到入侵特征的轉(zhuǎn)換邏輯Spark 這一層干的事是把原始日志行拆成字段然后按規(guī)則打標(biāo)簽。典型日志格式是 Nginx 或 Apache 的 combined 格式用正則提取 IP、時(shí)間、方法、路徑、狀態(tài)碼、UA。提取完之后做兩類判斷一類是閾值類比如同一 IP 在 60 秒內(nèi)請(qǐng)求超過(guò) 100 次標(biāo)記為brute_force另一類是模式類比如路徑里出現(xiàn)../或union select標(biāo)記為path_scan。# spark-job/log_analyzer.py 核心片段 from pyspark.sql import SparkSession from pyspark.sql.functions import regexp_extract, window, count, col spark SparkSession.builder \ .appName(LogIntrusionDetect) \ .config(spark.sql.shuffle.partitions, 4) \ .getOrCreate() # 從 Kafka 讀或從本地文件讀做離線驗(yàn)證 df spark.readStream.format(kafka) \ .option(kafka.bootstrap.servers, localhost:9092) \ .option(subscribe, access_log_topic) \ .load() # 提取字段正則按實(shí)際日志格式調(diào)整 parsed df.select( regexp_extract(col(value).cast(string), r^(\S), 1).alias(ip), regexp_extract(col(value).cast(string), r\[(.*?)\], 1).alias(ts), regexp_extract(col(value).cast(string), r(GET|POST) (\S), 2).alias(path), regexp_extract(col(value).cast(string), r (\d{3}) , 1).alias(status) ) # 60 秒窗口內(nèi)同 IP 請(qǐng)求計(jì)數(shù)超過(guò)閾值標(biāo)記 windowed parsed.groupBy( window(col(ts).cast(timestamp), 60 seconds), col(ip) ).count().filter(col(count) 100) query windowed.writeStream \ .outputMode(update) \ .format(console) \ .option(checkpointLocation, /tmp/spark_checkpoint) \ .start() query.awaitTermination()這段代碼的關(guān)鍵參數(shù)有三個(gè)spark.sql.shuffle.partitions控制聚合時(shí)的并行度本地跑設(shè) 4 就夠集群上按核數(shù)調(diào)window的60 seconds是滑動(dòng)窗口長(zhǎng)度改小會(huì)更靈敏但誤報(bào)多checkpointLocation必須指定否則流式作業(yè)重啟后無(wú)法恢復(fù)狀態(tài)。正則部分是最容易出問(wèn)題的地方不同日志格式字段順序不一樣建議先用head -5 access.log看一眼真實(shí)行再對(duì)著改正則。2.4 Flask 展示層把檢測(cè)結(jié)果變成能看的頁(yè)面Flask 這一層通常不直接連 Spark而是讀 Spark 寫出的結(jié)果表或 Redis 緩存。常見(jiàn)做法是 Spark 把告警寫入 MySQL 或 HiveFlask 用 SQLAlchemy 查出來(lái)渲染成表格和 ECharts 圖。如果你看到app.py里有pymysql或sqlalchemy的 import基本就是這個(gè)路子。# flask-web/app.py 核心片段 from flask import Flask, render_template from sqlalchemy import create_engine import pandas as pd app Flask(__name__) engine create_engine(mysqlpymysql://root:passwordlocalhost:3306/logdb?charsetutf8mb4) app.route(/) def index(): df pd.read_sql(SELECT ip, alert_type, COUNT(*) AS cnt FROM alerts GROUP BY ip, alert_type ORDER BY cnt DESC LIMIT 50, engine) return render_template(index.html, rowsdf.to_dict(records)) app.route(/api/alerts) def api_alerts(): df pd.read_sql(SELECT * FROM alerts ORDER BY ts DESC LIMIT 200, engine) return df.to_json(orientrecords, force_asciiFalse)這里create_engine的連接串要按你本地的 MySQL 賬號(hào)密碼改charsetutf8mb4不能省否則中文路徑會(huì)亂碼。/api/alerts是給前端 ECharts 異步拉數(shù)據(jù)用的返回 JSON 時(shí)force_asciiFalse保證中文可讀。如果 Flask 啟動(dòng)后頁(yè)面空白先看瀏覽器控制臺(tái)有沒(méi)有 500再看 MySQL 里alerts表是不是空的——Spark 沒(méi)寫進(jìn)去前端自然沒(méi)東西顯示。3. 從零跑通全鏈路環(huán)境準(zhǔn)備、啟動(dòng)順序與驗(yàn)證方法3.1 環(huán)境版本對(duì)齊JDK、Scala、Spark、Python 的兼容矩陣這套工程跑不起來(lái)十有八九是版本打架。Spark 3.x 默認(rèn)綁 Scala 2.12Spark 2.4 綁 Scala 2.11如果你下的包是 2.4 的卻裝了 2.12 的 Scala提交作業(yè)時(shí)會(huì)報(bào)NoSuchMethodError。Python 側(cè)PySpark 的版本必須和 Spark 本體一致pip install pyspark3.3.0就要配 Spark 3.3.0 的安裝包。組件推薦版本說(shuō)明JDK1.8 或 11Spark 3.x 建議 11Spark 2.4 只能 1.8Scala2.12.x與 Spark 3.x 對(duì)應(yīng)2.4 用 2.11Spark3.3.x穩(wěn)定且文檔多避免用 4.x 預(yù)覽版Python3.83.103.11 以上部分庫(kù)輪子不全Flume1.9 或 1.111.11 對(duì) TAILDIR 支持更好Flask2.x3.x 也可注意 Jinja2 語(yǔ)法差異對(duì)齊版本最省事的辦法是先spark-submit --version看輸出再python -c import pyspark; print(pyspark.__version__)兩個(gè)不一致就重裝。JDK 用java -version確認(rèn)如果是 17 而 Spark 是 2.4直接換 JDK 8別折騰參數(shù)。3.2 啟動(dòng)順序Flume → Kafka → Spark → Flask 的依賴鏈啟動(dòng)順序錯(cuò)了后面全白搭。正確鏈路是先起 Kafka如果 sink 用 Kafka再起 Flume 采集然后提交 Spark 流式作業(yè)最后起 Flask。因?yàn)?Spark 要訂閱 Kafka topictopic 不存在會(huì)直接報(bào)錯(cuò)退出Flask 要查 MySQL表沒(méi)建也會(huì) 500。# 1. 起 Kafka單機(jī)快速驗(yàn)證 bin/zookeeper-server-start.sh -daemon config/zookeeper.properties bin/kafka-server-start.sh -daemon config/server.properties bin/kafka-topics.sh --create --topic access_log_topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1 # 2. 起 Flume bin/flume-ng agent --conf conf --conf-file flume-conf/access-log.conf --name a1 -Dflume.root.loggerINFO,console # 3. 提交 Spark 流式作業(yè) spark-submit --master local[2] --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0 spark-job/log_analyzer.py # 4. 起 Flask cd flask-web python app.py每一步都有驗(yàn)證點(diǎn)Kafka 起完后jps應(yīng)該看到Kafka和QuorumPeerMainFlume 起完后往access.log追加一行控制臺(tái)應(yīng)打印出該行Spark 提交后控制臺(tái)應(yīng)出現(xiàn)Batch: 0之類的進(jìn)度Flask 起完后瀏覽器訪問(wèn)http://127.0.0.1:5000能看到頁(yè)面。哪一步?jīng)]輸出就停在那一步排查不要往下走。3.3 用模擬日志驗(yàn)證入侵檢測(cè)規(guī)則是否生效工程里一般帶一個(gè)logs/access.log或生成腳本。如果沒(méi)有自己造幾條能觸發(fā)規(guī)則的日志。比如同一 IP 連續(xù) 150 次請(qǐng)求或者路徑里帶../etc/passwd。追加日志用echo循環(huán)觀察 Spark 控制臺(tái)是否打出告警。# 模擬暴力破解同一 IP 快速請(qǐng)求 150 次 for i in $(seq 1 150); do echo 192.168.1.100 - - [10/Oct/2024:10:00:00 0800] GET /login HTTP/1.1 401 0 - curl/7.68 logs/access.log done # 模擬路徑掃描 echo 192.168.1.101 - - [10/Oct/2024:10:01:00 0800] GET /../../etc/passwd HTTP/1.1 404 0 - nikto logs/access.log追加后等一個(gè)窗口周期默認(rèn) 60 秒Spark 控制臺(tái)應(yīng)出現(xiàn)count 100的記錄Flask 頁(yè)面刷新后表格里應(yīng)出現(xiàn)192.168.1.100。如果沒(méi)出現(xiàn)先確認(rèn) Flume 是否真的讀到了新行看 Flume 日志再確認(rèn) Spark 的window時(shí)間字段解析是否正確——ts字段如果沒(méi)轉(zhuǎn)成 timestampwindow函數(shù)會(huì)直接報(bào)錯(cuò)或返回空。4. 避坑與排查這套工程最容易翻車的五個(gè)地方4.1 現(xiàn)象Flume 啟動(dòng)后日志不采集控制臺(tái)無(wú)輸出原因通常是TAILDIR的filegroups路徑寫錯(cuò)或者positionFile所在目錄沒(méi)有寫權(quán)限。Flume 對(duì)路徑錯(cuò)誤不敏感不會(huì)報(bào)致命錯(cuò)誤只是靜默不讀。解決方法是先用ls -l確認(rèn)日志文件存在且可讀再把positionFile指到/tmp下最后把 Flume 日志級(jí)別調(diào)到DEBUG看TaildirSource有沒(méi)有掃描到文件。4.2 現(xiàn)象Spark 提交報(bào)ClassNotFoundException: kafka.serializer.StringDecoder原因是--packages里的 Kafka 連接器版本和 Spark 版本不匹配。Spark 3.3 要用spark-sql-kafka-0-10_2.12:3.3.0如果寫成2.4.0就會(huì)找不到類。解決方法是先spark-submit --version確認(rèn) Spark 版本再把--packages的版本號(hào)改成一致。如果公司內(nèi)網(wǎng)拉不到包提前把 jar 下好放到$SPARK_HOME/jars下。4.3 現(xiàn)象Flask 頁(yè)面能打開(kāi)但表格為空MySQL 里也沒(méi)數(shù)據(jù)原因是 Spark 流式作業(yè)沒(méi)有把結(jié)果寫入 MySQL或者寫入了但表名不對(duì)。常見(jiàn)做法是 Spark 用foreachBatch寫 JDBC如果foreachBatch里沒(méi)調(diào)df.write.jdbc數(shù)據(jù)就只打在控制臺(tái)。解決方法是檢查 Spark 代碼里有沒(méi)有writeStream.foreachBatch或write.jdbc并確認(rèn) MySQL 的alerts表已建好字段和 DataFrame 的 schema 對(duì)得上。4.4 現(xiàn)象日志時(shí)間字段解析失敗window函數(shù)報(bào)AnalysisException原因是正則提取出的ts是字符串直接cast(timestamp)時(shí)格式不匹配。Nginx 默認(rèn)格式是10/Oct/2024:10:00:00 0800Spark 的to_timestamp默認(rèn)不認(rèn)這個(gè)格式。解決方法是顯式指定格式to_timestamp(col(ts), dd/MMM/yyyy:HH:mm:ss Z)注意MMM是英文月份縮寫本地化環(huán)境要設(shè)spark.sql.legacy.timeParserPolicyLEGACY。4.5 現(xiàn)象本地跑得好好的換臺(tái)機(jī)器就報(bào)No such file or directory: /tmp/spark_checkpoint原因是 checkpoint 路徑寫死在代碼里換機(jī)器后目錄不存在。Spark 流式作業(yè)的 checkpoint 目錄必須提前創(chuàng)建且要有寫權(quán)限。解決方法是在代碼里加os.makedirs(/tmp/spark_checkpoint, exist_okTrue)或者把路徑改成從環(huán)境變量讀部署時(shí)統(tǒng)一配。另外 checkpoint 目錄不要放在/tmp下長(zhǎng)期跑系統(tǒng)清理會(huì)把它刪掉導(dǎo)致作業(yè)恢復(fù)失敗。5. 進(jìn)階技巧把檢測(cè)規(guī)則從硬編碼改成可配置并用歷史日志回放驗(yàn)證5.1 規(guī)則外置用 JSON 配置替代寫死的閾值原始工程里閾值大概率是寫死在 Python 里的比如count 100。這樣改一次規(guī)則就要改代碼、重提交很麻煩。我一般會(huì)把規(guī)則抽成 JSONSpark 啟動(dòng)時(shí)讀一次廣播到各 executor。這樣調(diào)閾值不用動(dòng)代碼改完重啟作業(yè)即可。# rules.json { brute_force: {window_seconds: 60, threshold: 100, field: ip}, path_scan: {patterns: [../, union select, etc/passwd], field: path} }# 讀取規(guī)則并廣播 import json from pyspark.sql import SparkSession spark SparkSession.builder.appName(LogIntrusionDetect).getOrCreate() with open(rules.json, r, encodingutf-8) as f: rules json.load(f) bc_rules spark.sparkContext.broadcast(rules) # 在 foreachBatch 或 map 里用 bc_rules.value 取規(guī)則 threshold bc_rules.value[brute_force][threshold]廣播變量的好處是每個(gè) executor 只存一份不會(huì)因?yàn)橐?guī)則變大而拖慢序列化。參數(shù)上注意window_seconds和threshold要聯(lián)動(dòng)調(diào)窗口越長(zhǎng)閾值應(yīng)越高否則誤報(bào)會(huì)淹沒(méi)真實(shí)告警。patterns列表里的字符串會(huì)被拼成正則特殊字符要轉(zhuǎn)義比如../里的.要寫成\.。5.2 歷史日志回放用離線模式驗(yàn)證規(guī)則準(zhǔn)確率流式作業(yè)調(diào)試起來(lái)慢改一次等一個(gè)窗口。更高效的做法是先用離線模式跑歷史日志把規(guī)則調(diào)準(zhǔn)了再上流式。Spark 讀本地文件生成 DataFrame套用同樣的解析和判斷邏輯輸出告警數(shù)量和樣例人工看一眼誤報(bào)率。# 離線回放驗(yàn)證 df spark.read.text(logs/access.log) parsed df.select( regexp_extract(col(value), r^(\S), 1).alias(ip), regexp_extract(col(value), r\[(.*?)\], 1).alias(ts), regexp_extract(col(value), r(GET|POST) (\S), 2).alias(path) ) # 按 IP 聚合看哪些 IP 請(qǐng)求量最高 parsed.groupBy(ip).count().orderBy(col(count).desc()).show(10, truncateFalse) # 按路徑匹配可疑模式 suspicious parsed.filter(col(path).rlike((\\.\\./|union select|etc/passwd))) suspicious.show(20, truncateFalse)離線跑的好處是秒出結(jié)果不用等窗口。show(10, truncateFalse)不截?cái)嘧侄畏奖憧赐暾窂?。如果發(fā)現(xiàn)某個(gè)正常 IP 被誤判就把閾值調(diào)高或把該 IP 加白名單。白名單同樣可以放進(jìn)rules.json在過(guò)濾時(shí)filter(~col(ip).isin(whitelist))。5.3 一個(gè)我踩過(guò)的坑checkpoint 和規(guī)則變更的沖突有次我改了rules.json里的閾值重啟 Spark 作業(yè)后告警數(shù)量沒(méi)變。排查半天才發(fā)現(xiàn)流式作業(yè)的 checkpoint 里存了舊的查詢計(jì)劃規(guī)則雖然重新讀了但foreachBatch里用的還是廣播前的舊值。從那以后我每次改規(guī)則要么換一個(gè)新的checkpointLocation要么在代碼里加版本號(hào)規(guī)則版本變了就自動(dòng)切目錄。這個(gè)習(xí)慣幫我省了很多「改了沒(méi)生效」的玄學(xué)時(shí)間。import hashlib rule_hash hashlib.md5(json.dumps(rules, sort_keysTrue).encode()).hexdigest()[:8] checkpoint_path f/tmp/spark_checkpoint_{rule_hash}這樣規(guī)則一變checkpoint 目錄跟著變Spark 會(huì)當(dāng)成新作業(yè)啟動(dòng)不會(huì)復(fù)用舊狀態(tài)。代價(jià)是歷史狀態(tài)丟失但對(duì)入侵檢測(cè)這種場(chǎng)景重新開(kāi)始統(tǒng)計(jì)反而更干凈。希望幫到你。本文還有配套的精品資源點(diǎn)擊獲取