简介这是一套面向计算机相关专业学生与开发者的实时日志处理分析系统完整项目采用Spark、Flume、Kafka与HBase构建大数据流处理链路适合作为毕业设计、课程设计或大数据入门进阶的实战参考。资源包共85个文件约743KB以Java与Scala源码为主体辅以XML配置、properties参数文件、SQL脚本及JSP、HTML、JS前端页面并附有md说明文档与mvnw构建脚本覆盖数据采集、消息队列、流式计算到存储展示的完整模块。项目由专业团队开发源码经过测试功能稳定易复现目录结构清晰便于按模块理解系统架构与运行流程。已有71人学习关注具备一定基础的用户可在源码上二次修改扩展分析指标或对接其他数据源直接用于课题提交与项目演示初学者也可借此熟悉大数据组件的协同开发方式。1. 从一份毕设说起SparkFlumeKafkaHBase 实时日志链路到底解决什么问题很多同学做毕设时日志分析还停留在「把文件拷到本地用 Python 读一遍画几张图」的阶段。数据量一大、日志一实时产生这套做法立刻崩掉文件在滚动、进程在写、你读到的永远是半截。SparkFlumeKafkaHBase 这条链路本质上是把「日志产生 → 采集 → 缓冲 → 计算 → 存储 → 查询」拆成五段各司其职的流水线让实时日志处理从「事后批处理」变成「边产生边分析」。这套组合适合谁一是做大数据方向毕设、需要一条能跑通、能演示、能写进论文的完整链路的同学二是刚入行、想搞懂实时日志处理各组件边界在哪的工程师。它不追求极致性能但胜在组件成熟、资料多、每一段都能单独替换。下面我按「先立住原理、再动手复现、最后讲坑」的顺序把这条链路拆开讲清楚参数怎么设、失败看哪里都落到具体命令上。2. 链路选型为什么是 Flume 采、Kafka 缓、Spark 算、HBase 存2.1 四个组件各自的职责边界先把职责划清楚后面配置才不会互相打架。Flume 负责「从日志文件到消息队列」这一段它的核心是 Source、Channel、Sink 三件套擅长监听文件追加、做简单过滤和格式转换。Kafka 负责「缓冲和解耦」它把生产者Flume和消费者Spark在时间上彻底分开日志洪峰来了先堆在 Topic 里Spark 按自己的节奏消费不会因为计算慢就把采集端拖死。Spark 负责「计算」Structured Streaming 把流当成一张不断追加的表用 SQL 就能做窗口聚合、状态统计。HBase 负责「存储和随机查询」它基于 HDFS 但支持按 RowKey 毫秒级取单行适合存「按时间维度」组织的日志聚合结果。为什么不用 MySQL 存结果因为日志聚合结果按天、按小时、按维度累积行数增长快MySQL 单表几千万行就开始吃力而 HBase 天然水平扩展RowKey 设计好之后写入和查询都很稳。为什么不用 Flume 直接写 HBase因为中间少了 Kafka 这层缓冲Flume 的 Sink 一旦写 HBase 变慢Channel 会积压甚至丢数据实时日志处理最怕的就是采集端被下游拖垮。2.2 版本搭配与端口清单版本不匹配是这条链路翻车的第一大原因。我一般会锁定一套经过验证的组合避免用最新版互相踩。下面这套是常见且稳定的搭配组件建议版本关键端口用途Flume1.9.x44444Avro Source 接收Kafka2.8.x9092Broker 对外服务Spark3.2.x4040应用 Web UIHBase2.4.x16000Master 服务HBase2.4.x16020RegionServerZooKeeper3.6.x2181协调服务端口记不住没关系但要记住排查顺序先看 ZooKeeper 2181 通不通再看 Kafka 9092、HBase 16000。这三个端口任何一个不通链路都跑不起来。HBase 依赖 ZooKeeperKafka 也依赖 ZooKeeper所以 ZooKeeper 是整个链路的地基它一挂后面全乱。2.3 数据流方向与格式约定链路方向是单向的日志文件 → Flume Source → Flume Channel → Flume Sink → Kafka Topic → Spark Structured Streaming → HBase Table。每一段的数据格式要提前约定否则后面解析全是坑。我一般让 Flume 把每行日志原样发到 Kafka不做复杂解析把解析逻辑放到 Spark 里因为 Spark 的解析能力更强、改起来更方便。Kafka 里的消息就是一行字符串Spark 消费后按分隔符拆成字段再做聚合。提示格式约定越早定越好。日志里字段顺序、时间格式、分隔符一旦定下来Flume 拦截器、Spark 解析、HBase 列族设计都要跟着它走中途改格式等于全链路返工。3. 把链路跑起来Flume 采集到 Kafka 的最小配置3.1 Flume Agent 配置与启动命令先写 Flume 配置文件我一般命名为flume-kafka.conf放在 Flume 的 conf 目录下。核心是定义一个 Source 监听日志文件、一个 Kafka Channel 或 Memory Channel、一个 Kafka Sink。# flume-kafka.conf a1.sources r1 a1.channels c1 a1.sinks k1 # Source监听日志文件追加 a1.sources.r1.type exec a1.sources.r1.command tail -F /data/logs/app.log a1.sources.r1.channels c1 # Channel内存通道吞吐高但断电会丢毕设演示够用 a1.channels.c1.type memory a1.channels.c1.capacity 10000 a1.channels.c1.transactionCapacity 1000 # Sink写到 Kafka a1.sinks.k1.type org.apache.flume.sink.kafka.KafkaSink a1.sinks.k1.kafka.bootstrap.servers localhost:9092 a1.sinks.k1.kafka.topic log-topic a1.sinks.k1.flushSize 100 a1.sinks.k1.channels c1这段配置里exectail -F是最简单的文件监听方式适合日志持续追加的场景。capacity是 Channel 最多缓存多少条事件transactionCapacity是单次事务最多取多少条这两个值设太小会频繁触发事务、吞吐上不去设太大内存吃紧。KafkaSink 的flushSize是攒够多少条才批量发送设 100 到 500 之间比较平衡。启动命令bin/flume-ng agent \ --conf conf \ --conf-file conf/flume-kafka.conf \ --name a1 \ -Dflume.root.loggerINFO,console启动后如果控制台刷出Created topic log-topic或类似连接成功日志说明 Flume 已经连上 Kafka。如果卡在Connecting to Kafka先确认 Kafka 的 9092 端口是否监听、Topic 是否已创建。3.2 Kafka Topic 创建与消费验证Flume 启动前Topic 最好先手动建好避免自动创建时分区数不符合预期。# 创建 3 分区、1 副本的 Topic bin/kafka-topics.sh --create \ --bootstrap-server localhost:9092 \ --topic log-topic \ --partitions 3 \ --replication-factor 1 # 查看 Topic 详情 bin/kafka-topics.sh --describe \ --bootstrap-server localhost:9092 \ --topic log-topic分区数决定 Spark 消费的并行度毕设单机环境 3 个分区足够。副本数在单机只能是 1集群环境建议 2 或 3。验证 Flume 是否真的写进来了开一个控制台消费者bin/kafka-console-consumer.sh \ --bootstrap-server localhost:9092 \ --topic log-topic \ --from-beginning往/data/logs/app.log里追加几行控制台能实时打印出来说明「Flume → Kafka」这一段通了。这一步是整个链路最容易验证的环节务必先跑通再往下做。3.3 常见参数怎么调Flume 的batchSizeSink 端和 Kafka 的linger.ms、batch.size共同决定吞吐。日志量大时把 Flume Sink 的flushSize调到 500、Kafka Producer 的linger.ms设 10 到 50 毫秒能明显减少网络往返。但延迟敏感的场景要把linger.ms调小否则消息会在 Producer 端攒着不发。Kafka 消费端 Spark 的maxOffsetsPerTrigger控制每次触发最多拉多少条设太小吞吐低设太大单批处理时间长我一般按「每秒日志量 × 触发间隔」估算。4. Spark Structured Streaming 消费 Kafka 并写入 HBase4.1 消费 Kafka 的最小 Structured Streaming 程序Spark 3.x 用 Structured Streaming 消费 Kafka 非常简洁核心是readStream.format(kafka)加writeStream。下面是一个能跑通的最小示例做的是「按分钟统计日志条数」from pyspark.sql import SparkSession from pyspark.sql.functions import col, window, count, to_timestamp spark SparkSession.builder \ .appName(LogStreamToHBase) \ .config(spark.sql.shuffle.partitions, 3) \ .getOrCreate() # 从 Kafka 读取 df spark.readStream \ .format(kafka) \ .option(kafka.bootstrap.servers, localhost:9092) \ .option(subscribe, log-topic) \ .option(startingOffsets, latest) \ .load() # value 是二进制转成字符串再解析 lines df.selectExpr(CAST(value AS STRING) AS line) # 假设日志格式时间戳,级别,模块,消息 parsed lines.select( to_timestamp(col(line).substr(1, 19), yyyy-MM-dd HH:mm:ss).alias(ts), col(line).substr(21, 5).alias(level) ) # 按分钟窗口聚合 agg parsed.groupBy(window(col(ts), 1 minute), col(level)).agg(count(*).alias(cnt)) query agg.writeStream \ .outputMode(update) \ .foreachBatch(write_to_hbase) \ .option(checkpointLocation, /tmp/checkpoint/logstream) \ .trigger(processingTime10 seconds) \ .start() query.awaitTermination()startingOffsets设latest表示只消费新消息调试时想从头读改成earliest。trigger设 10 秒表示每 10 秒触发一批毕设演示够用。checkpointLocation必须设否则重启后 offset 丢失会重复消费。spark.sql.shuffle.partitions默认 200单机环境设成 3 到 10 能减少小文件。4.2 用 foreachBatch 写 HBase 的完整逻辑Structured Streaming 没有内置 HBase Sink标准做法是用foreachBatch拿到每批 DataFrame再用 HBase Java API 批量写。下面这段是核心写入函数import happybase def write_to_hbase(batch_df, batch_id): # 收集到 Driver 端数据量大时应改用 foreachPartition rows batch_df.collect() conn happybase.Connection(localhost, 9090) table conn.table(log_stats) with table.batch(batch_size100) as b: for row in rows: # RowKey级别 窗口开始时间保证同一维度聚集 rowkey f{row[level]}_{row[window][start].strftime(%Y%m%d%H%M)} b.put(rowkey, { binfo:level: row[level], binfo:cnt: str(row[cnt]), binfo:window_start: str(row[window][start]) }) conn.close()这里用happybase连接 HBase 的 Thrift 服务默认 9090 端口比直接用 Java API 写 Python 更省事。RowKey 设计成「级别_窗口时间」好处是同一级别的数据在 HBase 里物理相邻按级别扫描很快。batch_size100是攒批提交减少 RPC 次数。注意collect()会把整批数据拉到 Driver数据量大时要改成foreachPartition在 Executor 端直接写。4.3 HBase 表设计与建表命令HBase 表要在 Spark 任务启动前建好列族设计要贴合查询。日志聚合结果一般按「维度 时间」查所以列族不用多一个info就够。# 进入 HBase shell hbase shell # 建表表名 log_stats列族 info create log_stats, {NAME info, VERSIONS 1} # 查看表结构 describe log_stats # 扫描前 10 行验证 scan log_stats, {LIMIT 10}VERSIONS 1表示只保留一个版本日志统计结果不需要历史版本设大了反而占空间。如果要做「同一 RowKey 多次更新」VERSIONS设 1 时新值直接覆盖旧值符合统计场景。建表后可以用count log_stats看行数但大表慎用这个命令会全表扫描。注意HBase 的 RowKey 设计是整条链路性能的分水岭。RowKey 顺序写会热点集中加盐或哈希前缀能打散但会牺牲范围扫描能力。日志统计场景我一般用「维度前缀 时间」查询模式固定热点问题不严重。5. 避坑排查这条链路最容易翻车的 5 个地方5.1 Flume 报 ChannelFullException日志丢了一半现象Flume 控制台刷ChannelException: Space for commit to queue couldnt be acquiredKafka 里消息数明显少于日志行数。原因Sink 写 Kafka 的速度跟不上 Source 读文件的速度Memory Channel 的capacity被填满。解决先把capacity从 10000 调到 100000transactionCapacity调到 5000如果还满说明 Kafka 端确实慢检查 Kafka 是否单分区、Broker 是否磁盘 IO 打满。根治办法是换 File Channel牺牲一点吞吐换不丢数据。5.2 Spark 任务报 OffsetOutOfRangeException现象Spark 启动后立刻抛KafkaConsumer$OffsetOutOfRangeException任务起不来。原因startingOffsets设了earliest但 Kafka 里的消息已经因为保留策略被删了或者 checkpoint 里记录的 offset 超出了当前分区范围。解决删掉 checkpoint 目录重新跑或者把startingOffsets改成latest。生产环境要设auto.offset.reset为earliest并配合合理的retention.ms别让消息过期太快。5.3 HBase 写入报 RegionTooBusyException现象Spark 写 HBase 时偶发RegionTooBusyException: Over memstore limit。原因写入速度超过 RegionServer 的 memstore 刷盘速度单个 Region 的 memstore 超过阈值。解决调大hbase.hregion.memstore.flush.size默认 128MB或者给 RowKey 加随机前缀打散写入。毕设环境更简单的办法是降低 Spark 的写入频率把trigger从 10 秒调到 30 秒给 HBase 喘息时间。5.4 时间戳解析出来全是 null现象Spark 聚合结果里ts字段全是 null窗口聚合没输出。原因日志时间格式和to_timestamp的格式串不匹配比如日志是2024/01/01 10:00:00代码里写的是yyyy-MM-dd HH:mm:ss。解决先用lines.show(5, false)看原始行确认分隔符和位置再调整substr的起始位置和格式串。日志格式不规整时用正则regexp_extract比固定位置截取更稳。5.5 重启后数据重复统计现象Spark 任务重启后HBase 里同一窗口的计数翻倍。原因checkpoint 没设或设在了临时目录被清掉重启后从旧 offset 重新消费。解决checkpointLocation必须设在一个持久化路径且不要手动删。如果业务允许幂等可以在写 HBase 时用「窗口时间 维度」做 RowKey重复写会覆盖而不是累加天然去重。6. 进阶技巧让这条链路从「能跑」到「敢演示」毕设答辩最怕现场翻车所以最后一章讲几个让链路更稳的技巧。第一个是预置数据回放答辩前把一段历史日志用脚本按时间间隔逐行写入文件Flume 监听后链路自动跑起来比现场手动敲命令可靠得多。回放脚本很简单#!/bin/bash # replay.sh按行回放日志每行间隔 0.2 秒 while IFS read -r line; do echo $line /data/logs/app.log sleep 0.2 done /data/logs/sample.log这个脚本把sample.log里的历史日志按 0.2 秒一行追加到被 Flume 监听的文件里模拟实时产生。答辩时先跑回放再打开 Kafka 控制台消费者和 HBase shell数据一条条进来、表一行行长出来演示效果比静态截图强很多。第二个技巧是加一个查询验证环节。光写入不够要能查出来才算闭环。用 HBase shell 按 RowKey 前缀扫描# 查 ERROR 级别的统计结果 scan log_stats, {ROWFILTER PrefixFilter(ERROR_), LIMIT 20}PrefixFilter利用 RowKey 前缀做范围扫描比全表 scan 快得多。这也反过来验证了 RowKey 设计是否合理如果按级别查很慢说明 RowKey 前缀没设计对。第三个技巧是监控关键指标。Spark 的 4040 端口 Web UI 能看到每批的处理条数、延迟、积压答辩前打开这个页面用batchDuration和numInputRows两个指标说明链路是活的。Kafka 端用kafka-consumer-groups.sh --describe看 LagLag 持续为 0 说明消费跟得上。验证点命令/入口正常表现Flume 采集控制台日志无 ChannelFullExceptionKafka 积压kafka-consumer-groups.sh --describeLag 接近 0Spark 处理4040 Web UInumInputRows 持续增长HBase 写入scan log_stats行数随时间增加最后说个血泪经验这条链路我前后搭过不下十次每次翻车几乎都出在「版本不匹配」和「checkpoint 没设」这两件事上。版本锁定一套能跑的就别手痒去升级checkpoint 路径设好就别图省事删掉。把这两个习惯养成了SparkFlumeKafkaHBase 这条实时日志处理链路能稳到让你安心答辩。希望帮到你。本文还有配套的精品资源点击获取