使用Apache Spark实时分析网站点击流：检测DDoS攻击和异常流量排障复盘

业务场景

我们的业务场景是针对一个日访问量超过千万的高流量新闻网站，需要实时监控Nginx日志流，以快速识别和响应DDoS攻击。Nginx日志通过Kafka实时推送，数据格式为标准日志行，如"127.0.0.1 - - [10/Oct/2023:13:55:36 +0000] \"GET /news HTTP/1.1\" 200 1234"，包含IP、时间戳、请求方法、URL和状态码。团队规模为5人，包括数据分析师、大数据开发工程师和安全专家，协作边界是数据分析师负责规则定义，开发工程师实现Spark作业，安全专家验收效果。规模约束为处理峰值QPS 10万，要求检测延迟低于5秒，以防止服务中断。

方案对比决策

在确定实时分析方案时，我们围绕检测DDoS攻击的目标，比较了三种备选方案：Apache Spark Streaming、Apache Flink和基于Python脚本的简单阈值方法。

评估维度包括吞吐量、延迟、准确性、可维护性和资源消耗。Spark Streaming在吞吐量上表现优异，支持每秒处理数万条日志，延迟约2-5秒，社区成熟易于集成现有Hadoop栈；Flink延迟更低可达亚秒级，但团队熟悉度低，部署成本高；Python脚本方法部署快速，但扩展性差，无法处理高并发，易导致误报。

最终决策理由是选择Apache Spark Streaming，因为它平衡了性能、开发成本和团队技能，现有大数据平台基于Hadoop，迁移风险小。迁移策略是从当前使用的简单脚本系统逐步过渡，先灰度10%的流量到Spark作业，监控一周后无问题再全量切换，确保平滑升级。

里程碑计划

项目分为五个里程碑：需求分析（1周）、环境搭建与Kafka配置（2周）、Spark作业开发与测试（3周）、上线灰度（1周）、全量交付与复盘（1周）。验收口径为检测准确率（精确率和召回率）均需达到90%以上，平均延迟低于5秒，资源使用率不超过集群的30%。

实现步骤

数据说明：输入为Kafka中的Nginx日志流，每条日志解析出IP地址和时间戳，用于计算IP访问频率。任务类型为基于规则的异常检测，我们定义规则：如果一个IP在10秒内访问次数超过100次，则标记为疑似DDoS攻击，这是一个二分类问题（正常vs异常），指标匹配为使用精确率（Precision）和召回率（Recall）评估。

调用方式：通过spark-submit提交作业，命令示例如下：

spark-submit --class ClickStreamAnalysis --master yarn --executor-memory 4g --num-executors 5 streaming-job.jar

参数说明：–executor-memory设置每个执行器内存为4GB，–num-executors设置5个执行器以处理高负载，根据集群规模调整。

代码示例：以下为Scala代码片段，用于读取Kafka流并计算IP频率。

import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka010._

val ssc = new StreamingContext(sparkConf, Seconds(5))
val kafkaParams = Map("bootstrap.servers" -> "localhost:9092", "group.id" -> "click-stream")
val topics = Array("nginx-logs")
val stream = KafkaUtils.createDirectStream[String, String](
  ssc, LocationStrategies.PreferConsistent,
  ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
)

val ipCounts = stream.map(record => (record.value.split(" ").head, 1))
  .reduceByKeyAndWindow(_ + _, Seconds(10), Seconds(5))
  .filter { case (ip, count) => count > 100 }

ipCounts.print() // 输出疑似攻击IP

第二段代码为Python规则触发告警示例，集成到业务中：

import requests

# 当检测到异常时，调用API触发告警
def trigger_alert(ip, count):
    if count > 100:
        response = requests.post("http://alert-system/api/alert", json={"ip": ip, "count": count})
        print(f"Alert triggered for IP {ip} with count {count}")

验收标准

上线后评估基于两周运行数据：精确率达到92%，召回率89%，平均检测延迟为3秒，满足业务SLA。常见坑包括数据延迟导致误报（通过调整窗口大小为10秒缓解）、资源消耗过大（优化Spark配置如–executor-cores 2减少CPU使用）、误报触发不必要告警（添加IP白名单规则过滤正常爬虫）。

复盘总结

项目成功上线后，复盘显示Spark Streaming方案稳定处理了高流量，但初始阶段误报率较高，通过细化规则（如结合用户代理过滤）改善。团队协作顺畅，但建议未来引入自动化测试以加快迭代。下一步怎么接入业务：将检测结果实时推送到现有安全运营中心，自动触发防火墙规则以封锁异常IP。

角色分工与风险应对

角色分工中，数据分析师定义阈值规则，开发工程师编写和部署Spark作业，安全专家监控效果和调整策略。风险与应对包括数据丢失风险（通过Kafka副本和检查点机制备份）、资源瓶颈（设置集群自动扩缩容）、误报影响（建立反馈循环优化规则）。灰度策略为初始阶段只分析非核心页面日志，逐步扩展至全站。

交付清单

最终交付物包括Spark作业JAR包、Kafka配置文件、告警规则文档、性能指标报告和操作手册，确保团队可维护和扩展。