项目9：广告检测的流量作弊识别——Spark 综合实战

先修基础：项目1-8（Spark概述、Scala基础、Spark Shell编程、Spark IDE编程、Spark SQL、Spark Streaming、Spark GraphX、Spark MLlib）

第一部分：项目背景与需求分析
第二部分：常见流量作弊方式
第三部分：数据说明
第四部分：任务9.1-9.2 数据探索分析
第五部分：任务9.3 数据预处理与特征构建
第六部分：特征标准化与数据划分
第七部分：任务9.4 逻辑回归模型构建与评估
第八部分：随机森林模型构建与评估
第九部分：模型加载与应用
第十部分：完整项目流程总结
第十一部分：常见问题与排错指南
第十二部分：实习 / 面试高频考点

第一部分：项目背景与需求分析

1.1 项目背景

什么是虚假流量？

互联网虚假流量，是指通过特殊方式模仿人类浏览行为生成的访问流量。比如：写个程序，每分钟自动访问一次某网站主页，这就是虚假流量。

为什么会有虚假流量？

浏览量增加 → 广告费用更高
部分网站受利益驱使 → 通过作弊产生虚假流量
给广告主带来严重损失（花了钱没效果）

本项目要做什么？

用 Spark 大数据技术实现广告流量作弊识别，判断哪些访问是真实的，哪些是作弊的。

💡 项目定位：这是一个综合实战项目，把前面学的 Spark SQL、MLlib、Hive 等知识全部串起来用。模拟真实生产环境中的完整流程：数据探索 → 特征工程 → 模型训练 → 模型评估 → 模型上线。

1.2 需求分析

实现流程

历史流量数据 → 数据抽取和划分 → 探索分析 → 数据预处理 → 特征构建 → 建模 → 评估 → 保存模型 → 应用

三大核心任务

任务	内容
任务1	探索分析广告流量数据
任务2	预处理数据并构建特征
任务3	构建与评估分类模型

第二部分：常见流量作弊方式

2.1 三种常见作弊方式

方式一：脚本刷新网页作弊

做法：写个脚本，自动刷新网页
特征：cookie 和 ip 都不变，但访问次数特别多
识别方法：统计同一 ip + cookie 的访问次数，超过阈值就是作弊

做法：每次刷完就清除 cookie，制造新的 cookie
特征：ip 不变，但 cookie 一直在变（一个 ip 对应很多不同 cookie）
识别方法：统计每个 ip 对应的不同 cookie 数量

方式三：ADSL 重新拨号后刷新网页作弊

做法：ADSL 拨号上网，每次重新拨号 ip 就变，但 ip 都来自同一个区域
特征：ip 的前两段或前三段相同（同网段）
识别方法：统计 ip 前两段/前三段相同的记录数

三种作弊方式对比

作弊方式	核心特征	对应特征名
脚本刷新	同一 ip + cookie 访问次数多	N
清除 cookie	同一 ip 对应多个 cookie	N1
ADSL 拨号	ip 前两段/前三段相同的记录多	N2 / N3

💡 记忆口诀：
N = 同一 ip+cookie 次数（最直接的作弊）
N1 = 同一 ip 的 cookie 数（换 cookie 作弊）
N2 = ip 前两段相同数（大范围拨号作弊）
N3 = ip 前三段相同数（小范围拨号作弊）

第三部分：数据说明

3.1 数据文件

文件名：case_data_new.csv

7天的广告流量数据
共 1,704,154 条记录（约170万条）
22个字段

3.2 主要字段

字段	说明
`dt`	相对天数（1~7）
`ip`	访问者IP地址
`cookie`	浏览器cookie
`timestamps`	时间戳
`label`	是否作弊（标签列，0=正常，1=作弊）
`idfa`	iOS设备标识
`imei`	Android设备标识
`android`	Android ID
`openudid`	OpenUDID
`mac`	MAC地址
`creativeid`	广告创意ID
`mobile_os`	手机操作系统
`mobile_type`	手机型号
`app_key_md5`	App密钥MD5
`app_name_md5`	App名称MD5
`os_type`	操作系统类型
`useragent`	用户代理
...	其他字段

⚠️ 注意：大部分字段是字符类型，不能直接用于数值型的分类模型，需要做编码或构建新特征。

第四部分：任务9.1-9.2 数据探索分析

4.1 任务目标

对广告流量数据进行全面探索：

基础探索：记录数、日流量、数据类型、缺失值、冗余数据
业务探索：三种作弊方式的数据特征

4.2 基础探索

步骤1：创建 SparkSession

scala

import org.apache.spark.sql.SparkSession

object Explore {
  def main(args: Array[String]): Unit = {
    // 创建SparkSession（本地模式）
    val spark = SparkSession.builder()
      .appName("Explore")
      .master("local[*]")
      .enableHiveSupport()
      .getOrCreate()
    
    // 设置日志级别
    spark.sparkContext.setLogLevel("WARN")
  }
}

步骤2：读取数据并统计记录数

scala

// 读取CSV数据
val rawData = spark.read
  .option("header", "true")
  .csv("E:\\data\\case_data_new.csv")

// 统计数据记录数
println("原始数据集行数为：" + rawData.count())

结果：1,704,154 条（约170万条）

步骤3：统计日流量

scala

// 按dt分组统计每天的数据量
rawData.groupBy("dt")
  .count()
  .selectExpr("dt", "count as dayCount")
  .sort("dt")
  .show()

结果分析：

7天的数据，每天的流量差异不大
数据产生环境相对稳定
不存在数据倾斜问题

步骤4：分析数据类型

大部分变量是字符类型（String），不适合直接用于数值型分类模型。后续需要：

对字符类型数据进行编码
或者根据字符类型字段构造新的特征列

步骤5：统计缺失数据

自定义缺失率计算函数

scala

import org.apache.spark.sql.DataFrame

// 自定义函数：统计缺失值比率
def MissingCount(data: DataFrame, columnName: String): Unit = {
  var missingRate = 0.0
  
  if (columnName != "creativeid") {
    // 普通字段：空值算缺失
    missingRate = 1 - (data.select(columnName).na.drop().count().toDouble / data.count())
  } else {
    // creativeid特殊：值为0也算缺失
    missingRate = data.select(columnName)
      .filter("creativeid == 0")
      .count() / data.count().toDouble
  }
  
  // 只打印有缺失的字段
  if (missingRate > 0) {
    println(columnName + " 缺少值比率：" + missingRate * 100 + "%")
  }
}

调用函数统计所有字段

scala

// 获取所有列名
val columnName = rawData.columns.toList

// 遍历每个字段，统计缺失率
for (i <- columnName) {
  MissingCount(rawData, i)
}

结果分析：

22个字段中，有12个字段存在缺失值
creativeid 缺失率高达 98.39%（值为0的视为缺失）
很多字段缺失率很高，而且是字符型，无法插补
后续需要删除缺失率过高的字段

⚠️ 为什么 creativeid 为 0 算缺失？ 正常情况下 creativeid 应该是有效的广告创意ID，大量为0不符合正常情况，所以把0值也视为缺失。

步骤6：冗余数据分析

发现：idfa、imei、android、openudid 这4个字段：

缺失率都偏高
都是用于识别手机系统类型的字段
功能重复，属于冗余字段

处理思路：

如果后续需要手机系统类型特征，可以将这4个字段合并
降低缺失率，提取有效信息

4.3 业务探索：作弊特征分析

探索一：脚本刷新网页作弊

思路：同一 ip + cookie 的访问次数特别多 → 可能是脚本刷的

scala

import org.apache.spark.sql.functions.{col, desc}

// 按ip和cookie分组，统计每个组合的访问次数
val cookie_ip_distribute = rawData.groupBy("ip", "cookie")
  .count()
  .withColumn("ip_cookie_count_precent",
    col("count") / rawData.count() * 100)  // 计算占比
  .orderBy(desc("count"))  // 按次数降序

cookie_ip_distribute.show(false)

// 统计超过100次的记录数
val click_gt_100 = cookie_ip_distribute.filter("count > 100").count()
println("同ip、cookie出现超过100次以上的记录数：" + click_gt_100)

结果：

确实存在同一 ip + cookie 高频浏览的情况
超过100次的有 104 个
这些占据了大量流量记录

💡 为什么选100次作为阈值？ 正常情况下，一个人7天内频繁点击同一个广告100次以上是不太正常的。这个阈值是根据业务经验设定的。

思路：同一 ip 对应很多不同的 cookie → 可能是定期清 cookie 刷的

scala

import org.apache.spark.sql.functions.countDistinct

// 按ip分组，统计每个ip对应的不同cookie数量
val ip_distribute = rawData.groupBy("ip")
  .agg(countDistinct("cookie") as "ip_count")  // 每个ip有多少个不同cookie
  .groupBy("ip_count")
  .agg(
    count("ip_count") as "ip_count_count",        // 有多少个ip是这个cookie数
    count("ip_count") / rawData.count() * 100 as "ip_count_count_precent"  // 占比
  )
  .orderBy(desc("ip_count"))

ip_distribute.show(false)

结果：

存在同一个 ip 高频访问广告的情况
占据较高的数据比例
需要进行识别

探索三：ADSL 重新拨号作弊

思路：ip 前两段/前三段相同的记录特别多 → 可能是同一区域拨号刷的

按 ip 前两段统计

scala

import org.apache.spark.sql.functions.substring_index

// 提取ip前两段
val ip_two = rawData.withColumn("ip_two",
    substring_index(col("ip"), ".", 2))
  .groupBy("ip_two")
  .agg(count("ip_two") as "ip_two_count")
  .orderBy(desc("ip_two_count"))

ip_two.show(false)

按 ip 前三段统计

scala

// 提取ip前三段
val ip_three = rawData.withColumn("ip_three",
    substring_index(col("ip"), ".", 3))
  .groupBy("ip_three")
  .agg(count("ip_three") as "ip_three_count")
  .orderBy(desc("ip_three_count"))

ip_three.show(false)

结果：

ip 前两段相同的流量记录数在10000以上的情况较多
ip 前三段相同的流量记录上万的情况也较多
如果对应的记录数过于庞大，可大致判定为虚假流量

💡 substring_index 函数： substring_index(col("ip"), ".", 2) 表示按"."分割，取前2段。比如 "192.168.1.100" → "192.168"

4.4 探索分析小结

探索维度	发现	处理方式
数据量	170万条，7天	数据量较大，用集群模式
日流量	每天差异不大	无数据倾斜
数据类型	大多数字符型	需要编码或构建新特征
缺失值	12个字段有缺失，部分高达98%	删除缺失率过高的字段
冗余数据	4个设备标识字段功能重复	后续可合并
脚本作弊	同一ip+cookie超100次的有104个	构建特征N
清cookie作弊	同一ip对应多个cookie	构建特征N1
ADSL作弊	ip前两段/前三段相同的记录多	构建特征N2、N3

第五部分：任务9.3 数据预处理与特征构建

5.1 数据预处理

删除缺失率过高的字段

删除以下7个缺失率过高的字段：

mac
creativeid
mobile_os
mobile_type
app_key_md5
app_name_md5
os_type

scala

// 删除缺失率过高的字段
val data_new = rawData
  .drop("mac")
  .drop("creativeid")
  .drop("mobile_os")
  .drop("mobile_type")
  .drop("app_key_md5")
  .drop("app_name_md5")
  .drop("os_type")

⚠️ 注意： idfa、imei、android、openudid 这4个字段暂时保留，因为后续构建特征时不确定是否需要用到。

保存到 Hive

准备工作

启动 Hadoop 集群
启动 Hive 元数据服务
在 Hive CLI 中创建数据库：create database ad_traffic;
把 hive-site.xml 复制到项目的 resources 目录

保存数据

scala

// 保存处理后的数据到Hive表（覆盖模式）
data_new.write
  .mode("overwrite")
  .saveAsTable("ad_traffic.AdData")

保存的表：ad_traffic.AdData

5.2 特征构建

构建哪几个特征？

根据前面的探索分析，构建4个特征：

特征名	含义	对应作弊方式
N	同一 ip + cookie 的访问次数	脚本刷新作弊
N1	同一 ip 对应的不同 cookie 数量	清除 cookie 作弊
N2	ip 前两段相同的记录数	ADSL 拨号作弊（大范围）
N3	ip 前三段相同的记录数	ADSL 拨号作弊（小范围）

为什么按时间区间构建特征？

作弊行为是有时间性的
用5小时作为一个时间窗口（根据广告点击周期的频率划分）
每个时间区间内单独计算特征
这样能更好地捕捉短期作弊行为

时间区间划分

scala

import org.apache.spark.sql.functions.{col, max, min}
import org.apache.spark.sql.types.DataTypes

// 读取预处理后的数据
val data = spark.read.table("ad_traffic.AdData")

// 计算最大时间戳和最小时间戳
val max_min_timestamp = data.select(
    max(col(timestamps).cast(DataTypes.IntegerType)) as "max_ts",
    min(col(timestamps).cast(DataTypes.IntegerType)) as "min_ts"
  ).rdd.collect()

val max_ts = max_min_timestamp(0).getInt(0)
val min_ts = max_min_timestamp(0).getInt(1)

// 以18000秒（5小时）为间隔切割时间段
val times = List.range(min_ts, max_ts, 18000)
println("时间分割点：" + times)

💡 List.range 用法： List.range(start, end, step) 从 start 到 end（不包含end），步长为 step。比如 List.range(0, 10, 3) → List(0, 3, 6, 9)

循环构建特征

scala

import org.apache.spark.sql.functions.{count, countDistinct, substring_index}

// 定义列名
val timestamps = "timestamps"
val cookie = "cookie"
val ip = "ip"
val N = "N"
val N1 = "N1"
val N2 = "N2"
val N3 = "N3"
val ranks = "rank"

// 获取前5个时间段（25小时）的特征
for (i <- 0 to 4) {
  // 1. 筛选当前时间区间的数据
  val data_sub = data.filter(
    "timestamps>=" + times(i) + " and timestamps<" + times(i + 1))
  
  // 2. 构建特征N：同一ip+cookie的访问次数
  val data_N_sub = data_sub
    .groupBy(cookie, ip)
    .agg(count(ip) as N)
    .join(data_sub, Seq(cookie, ip), "inner")
    .select(ranks, N)
  
  // 3. 构建特征N1：同一ip对应的不同cookie数量
  val data_N1_sub = data_sub
    .groupBy(ip)
    .agg(countDistinct(cookie) as N1)
    .join(data_sub, Seq(ip), "inner")
    .select(ranks, N1)
  
  // 4. 构建特征N2：ip前两段相同的记录数
  val data_ip_two = data_sub
    .withColumn("ip_two", substring_index(col(ip), ".", 2))
  val data_N2_sub = data_ip_two
    .groupBy("ip_two")
    .agg(count("ip_two") as "N2")
    .join(data_ip_two, Seq("ip_two"), "inner")
    .select(ranks, N2)
  
  // 5. 构建特征N3：ip前三段相同的记录数
  val data_ip_three = data_sub
    .withColumn("ip_three", substring_index(col(ip), ".", 3))
  val data_N3_sub = data_ip_three
    .groupBy("ip_three")
    .agg(count("ip_three") as "N3")
    .join(data_ip_three, Seq("ip_three"), "inner")
    .select(ranks, N3)
  
  // 6. 合并4个特征并保存到Hive表（追加模式）
  val data_model_N = data_N_sub
    .join(data_N1_sub, ranks)
    .join(data_N2_sub, ranks)
    .join(data_N3_sub, ranks)
  
  data_model_N.write
    .mode("append")
    .saveAsTable("ad_traffic.TimeFeatures")
}

代码说明

代码	说明
`for (i <- 0 to 4)`	处理前5个时间区间（25小时）
`data_sub`	当前时间区间的数据
`groupBy + count`	统计访问次数
`countDistinct`	统计去重后的数量
`join(...).select(ranks, N)`	把统计结果关联回原数据，每条记录都带上特征值
`mode("append")`	追加模式，每个区间的结果累加到同一张表

⚠️ 为什么只取前25小时？ 7天的数据量非常大，构建特征对硬件要求很高，会耗费大量时间。为了演示方便，取前25小时的数据。实际生产环境中会处理全部数据。

💡 为什么用 ranks 字段关联？ranks（或 rank）是每条记录的唯一标识，通过它把4个特征合并到同一条记录上。

合并特征和标签

scala

// 读取特征表和原始数据表，通过ranks字段连接
val FeaturesData = spark.read.table("ad_traffic.TimeFeatures")
  .join(data, ranks)
  .select(
    col(ranks),
    col("dt"),
    col("N"),
    col("N1"),
    col("N2"),
    col("N3"),
    col("label").cast("double")  // 标签转成double类型
  )

// 保存到Hive
FeaturesData.write
  .mode("overwrite")
  .saveAsTable("ad_traffic.FeaturesData")

保存的表：ad_traffic.FeaturesData

字段：rank, dt, N, N1, N2, N3, label

⚠️ 注意： TimeFeatures 表只有特征字段，没有 label 字段。 label 在 AdData 表中。需要 join 把 label 加进来。

第六部分：特征标准化与数据划分

6.1 为什么要标准化？

4个特征的值差异很大：

N：可能是几十、几百
N1：可能是几个、几十个
N2、N3：可能是几千、几万

如果不标准化，数值大的特征会对模型产生更大的影响，这不公平。

6.2 使用 MinMaxScaler 归一化

选择 MinMaxScaler（最小-最大归一化），把特征缩放到 [0,1] 区间。

scala

import org.apache.spark.ml.feature.{MinMaxScaler, VectorAssembler}

object Scaler {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("Scaler")
      .master("local[*]")
      .enableHiveSupport()
      .getOrCreate()
    spark.sparkContext.setLogLevel("WARN")
    
    // 1. 读取特征数据
    val data = spark.read.table("ad_traffic.FeaturesData")
    
    // 2. 特征向量化：把4个特征合并成一个向量
    val VectorData = new VectorAssembler()
      .setInputCols(Array("N", "N1", "N2", "N3"))
      .setOutputCol("VectorFeatures")
      .transform(data)
    
    // 3. MinMaxScaler归一化
    val MaxMin = new MinMaxScaler()
      .setInputCol("VectorFeatures")
      .setOutputCol("features")
      .fit(VectorData)
    
    val dataScaler = MaxMin.transform(VectorData)
    
    // 4. 划分数据集（7:3）
    val Array(modelData, loadData) = dataScaler.randomSplit(Array(0.7, 0.3))
    
    // 5. 保存到Hive
    modelData.write.mode("overwrite").saveAsTable("ad_traffic.ModelData")
    loadData.write.mode("overwrite").saveAsTable("ad_traffic.LoadData")
    
    // 6. 打印数据量
    println("数据集总数据量：" + dataScaler.count())
    println("模型构建数据量：" + modelData.count())
    println("模型加载数据量：" + loadData.count())
  }
}

数据划分说明

数据集	比例	用途	表名
modelData	70%	模型构建与评估	ad_traffic.ModelData
loadData	30%	模拟真实模型应用	ad_traffic.LoadData

结果：

总数据量：702,135
模型构建数据：491,698
模型加载数据：210,437

💡 为什么分两份？
modelData：用来训练和测试模型（内部再按7:3分训练集和测试集）
loadData：模拟"新数据"，用来测试保存后的模型加载和预测效果这样可以验证模型保存和加载后效果是否一致

第七部分：任务9.4 逻辑回归模型构建与评估

7.1 为什么用逻辑回归？

这是一个二分类问题（是作弊 / 不是作弊）
逻辑回归是二分类的经典算法
原理简单，准确率较高
适合作为基准模型

7.2 代码实现

scala

import org.apache.spark.ml.classification.LogisticRegression
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.sql.SparkSession

object Logistic {
  def main(args: Array[String]): Unit = {
    // 1. 创建SparkSession（集群模式！）
    val spark = SparkSession.builder()
      .appName("Logistic")
      .master("spark://master:7077")  // 独立集群模式
      .enableHiveSupport()
      .getOrCreate()
    spark.sparkContext.setLogLevel("WARN")
    
    // 2. 从命令行参数获取输入表和输出路径
    val inputTable = args(0)
    val output = args(1)
    
    // 3. 读取数据
    val ModelData = spark.read.table(inputTable)
    
    // 4. 划分训练集和测试集（7:3）
    val Array(train, test) = ModelData.randomSplit(Array(0.7, 0.3))
    
    // 5. 构建逻辑回归模型
    val model = new LogisticRegression()
      .setElasticNetParam(0.03)   // 弹性网络正则化参数
      .setMaxIter(15)             // 最大迭代次数
      .fit(train)                 // 训练
    
    // 6. 预测
    val pre = model.transform(test)
    pre.select("label", "prediction").show()
    
    // 7. 评估（准确率）
    val evaluator = new MulticlassClassificationEvaluator()
      .setLabelCol("label")
      .setPredictionCol("prediction")
      .setMetricName("accuracy")
    
    println("Logistic Model Accuracy：" + evaluator.evaluate(pre))
    
    // 8. 保存模型
    model.write.overwrite().save(output)
  }
}

代码说明

代码	说明
`.master("spark://master:7077")`	独立集群模式，不是本地模式
`args(0)`	第一个命令行参数：输入表名
`args(1)`	第二个命令行参数：模型保存路径
`setElasticNetParam(0.03)`	弹性网络混合参数，防止过拟合
`setMaxIter(15)`	最大迭代15次

⚠️ 注意：逻辑回归算法会自动寻找列名为 features 的特征列和列名为 label 的标签列，所以不需要额外设置 setFeaturesCol 和 setLabelCol。前提是你的列名就叫 features 和 label！

7.3 提交到集群运行

打包 JAR

在 IDEA 中 Build Artifacts，打包成 model.jar

上传到 Linux

bash

# 上传到 /opt/data 目录

spark-submit 提交

bash

spark-submit --master spark://master:7077 \
  --class Logistic /opt/data/model.jar \
  ad_traffic.ModelData \
  /tipdm/data/model/Logistic

参数说明

参数	说明
`--master spark://master:7077`	提交到独立集群
`--class Logistic`	主类名
`ad_traffic.ModelData`	args(0)：输入表名
`/tipdm/data/model/Logistic`	args(1)：模型保存路径（HDFS）

7.4 模型效果

准确率：约 85.8%

对于大部分作弊访问记录都可以识别
效果还可以，但还有提升空间

第八部分：随机森林模型构建与评估

8.1 为什么用随机森林？

逻辑回归准确率 85.8%，还有提升空间
随机森林在数据足够的情况下，准确率通常更高
可以对比两种算法的效果

8.2 代码实现

scala

import org.apache.spark.ml.classification.RandomForestClassifier
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.sql.SparkSession

object RandomForest {
  def main(args: Array[String]): Unit = {
    // 1. 创建SparkSession（集群模式）
    val spark = SparkSession.builder()
      .appName("RandomForest")
      .master("spark://master:7077")
      .enableHiveSupport()
      .getOrCreate()
    spark.sparkContext.setLogLevel("WARN")
    
    // 2. 命令行参数
    val inputTable = args(0)
    val output = args(1)
    
    // 3. 读取数据
    val ModelData = spark.read.table(inputTable)
    
    // 4. 划分训练集和测试集
    val Array(train, test) = ModelData.randomSplit(Array(0.7, 0.3))
    
    // 5. 构建随机森林分类模型
    val rfModel = new RandomForestClassifier()
      .setFeaturesCol("features")
      .setLabelCol("label")
      .setNumTrees(5)  // 5棵决策树
      .fit(train)
    
    // 6. 预测
    val pre = rfModel.transform(test)
    pre.select("label", "prediction").show()
    
    // 7. 评估
    val evaluator = new MulticlassClassificationEvaluator()
      .setLabelCol("label")
      .setPredictionCol("prediction")
      .setMetricName("accuracy")
    
    println("RandomForest Model Accuracy：" + evaluator.evaluate(pre))
    
    // 8. 保存模型
    rfModel.write.overwrite().save(output)
  }
}

8.3 提交到集群运行

bash

spark-submit --master spark://master:7077 \
  --class RandomForest /opt/data/model.jar \
  ad_traffic.ModelData \
  /tipdm/data/model/RandomForest

8.4 模型效果对比

模型	准确率	特点
逻辑回归	约 85.8%	简单、快速、可解释性强
随机森林	约 91.1%	准确率更高，提升约5%

为什么随机森林更好？

逻辑回归本质是在线性回归上加 sigmoid 函数
容易出现梯度消失的情况
随机森林是多棵树投票，泛化能力更强
在数据足够的情况下，随机森林准确率更高

💡 实际工作中的选择：
先跑逻辑回归作为 baseline（基准）
再跑随机森林、GBDT 等更复杂的模型
对比效果，选最好的
还要考虑训练速度、可解释性等因素

第九部分：模型加载与应用

9.1 为什么要加载模型？

训练好的模型保存起来后，实际使用时需要：

加载保存好的模型
对新数据进行预测
验证模型加载后效果是否一致

9.2 代码实现

scala

import org.apache.spark.ml.classification.{LogisticRegressionModel, RandomForestClassificationModel}
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.sql.SparkSession

object LoadModel {
  def main(args: Array[String]): Unit = {
    // 1. 创建SparkSession（本地模式）
    val spark = SparkSession.builder()
      .appName("LoadModel")
      .master("local[*]")
      .enableHiveSupport()
      .getOrCreate()
    spark.sparkContext.setLogLevel("WARN")
    
    // 2. 读取测试数据（模拟新数据）
    val LoadData = spark.read.table("ad_traffic.LoadData")
    
    // 3. 加载逻辑回归模型
    val LogisticModel = LogisticRegressionModel
      .load("hdfs://master:8020/tipdm/data/model/Logistic")
    
    // 4. 逻辑回归预测 + 评估
    val LogisticPre = LogisticModel.transform(LoadData)
    val LogisticAcc = new MulticlassClassificationEvaluator()
      .setLabelCol("label")
      .setPredictionCol("prediction")
      .setMetricName("accuracy")
      .evaluate(LogisticPre)
    println("逻辑回归模型后期数据准确率：" + LogisticAcc)
    
    // 5. 加载随机森林模型
    val RandomForest = RandomForestClassificationModel
      .load("hdfs://master:8020/tipdm/data/model/RandomForest")
    
    // 6. 随机森林预测 + 评估
    val RandomForestPre = RandomForest.transform(LoadData)
    val RandomForestAcc = new MulticlassClassificationEvaluator()
      .setLabelCol("label")
      .setPredictionCol("prediction")
      .setMetricName("accuracy")
      .evaluate(RandomForestPre)
    println("随机森林模型后期数据准确率：" + RandomForestAcc)
  }
}

代码说明

代码	说明
`LogisticRegressionModel.load(path)`	加载逻辑回归模型
`RandomForestClassificationModel.load(path)`	加载随机森林模型
`hdfs://master:8020/...`	模型保存在HDFS上，路径要写完整

⚠️ 注意模型类名：
训练时用的是 LogisticRegression（算法类）
加载时用的是 LogisticRegressionModel（模型类）
随机森林同理：RandomForestClassifier → RandomForestClassificationModel 别搞混了！

9.3 结果说明

保存的模型再次加载后，效果不会损失
真实场景中，新数据是没有 label 的
这里因为是模拟，数据有 label，所以可以计算准确率验证效果

💡 真实生产环境的模型应用流程：
实时/离线采集新数据
做同样的特征工程（和训练时一致）
加载模型
预测
输出结果（比如标记为作弊流量）

第十部分：完整项目流程总结

10.1 整体流程图

┌─────────────────────────────────────────────────────────────┐
│                    广告流量作弊识别项目                        │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
                    ┌─────────────────┐
                    │   原始数据        │  case_data_new.csv
                    │  (170万条, 22字段)│
                    └────────┬────────┘
                             │
                             ▼
                    ┌─────────────────┐
                    │  数据探索分析     │  Explore.scala
                    │  （本地模式）     │
                    └────────┬────────┘
                             │
              ┌──────────────┼──────────────┐
              ▼              ▼              ▼
        基础探索         缺失值分析       作弊特征探索
        (记录数/日流量)  (12个字段有缺失)  (3种作弊方式)
                             │
                             ▼
                    ┌─────────────────┐
                    │  数据预处理       │  删除7个高缺失字段
                    │                 │  保存到Hive: AdData
                    └────────┬────────┘
                             │
                             ▼
                    ┌─────────────────┐
                    │  特征构建         │  Features.scala
                    │  （本地模式）     │  N, N1, N2, N3
                    │                 │  保存到Hive: FeaturesData
                    └────────┬────────┘
                             │
                             ▼
                    ┌─────────────────┐
                    │  特征标准化       │  Scaler.scala
                    │  （本地模式）     │  MinMaxScaler
                    │                 │  7:3划分 → ModelData + LoadData
                    └────────┬────────┘
                             │
              ┌──────────────┴──────────────┐
              ▼                             ▼
    ┌─────────────────┐           ┌─────────────────┐
    │  逻辑回归模型     │           │  随机森林模型     │
    │  （集群模式）     │           │  （集群模式）     │
    │  Logistic.scala │           │ RandomForest.scala
    │  准确率: 85.8%  │           │  准确率: 91.1%  │
    └────────┬────────┘           └────────┬────────┘
              │                             │
              └──────────────┬──────────────┘
                             ▼
                    ┌─────────────────┐
                    │  模型加载与应用   │  LoadModel.scala
                    │  （本地模式）     │  模拟新数据预测
                    └─────────────────┘

10.2 各阶段使用的运行模式

阶段	运行模式	原因
数据探索	本地模式	探索阶段，数据量可以抽样
特征构建	本地模式	演示用，只取前25小时
特征标准化	本地模式	数据量已减小
模型训练	集群模式	正式训练，需要集群算力
模型加载	本地模式	预测阶段，数据量小

💡 实际生产环境：所有阶段都应该在集群上运行，特别是数据量大的时候。这里为了教学方便，探索和特征构建用本地模式。

10.3 用到的 Spark 技术

技术	用途	对应项目
Spark SQL / DataFrame	数据读取、探索、处理	项目5
Hive 集成	数据存储（saveAsTable）	项目5
Spark MLlib	特征工程、模型训练、评估	项目8
spark-submit	集群提交任务	项目4

第十一部分：常见问题与排错指南

11.1 数据探索类

问题1：substring_index 不生效

现象：ip 分段不对原因：分隔符写错了，或者参数不对解决：

第二个参数是分隔符："."
第三个参数是取前几段：2 表示前两段
substring_index(col("ip"), ".", 2) → "192.168.1.100" → "192.168"

问题2：缺失率计算不对

现象：算出来的缺失率是0或100% 原因：

字符类型的空值可能是空字符串，不是 null
na.drop() 只删 null，不删空字符串解决：
先把空字符串转成 null：when(col(c) === "", null).otherwise(col(c))
或者过滤空字符串：filter(col(c) =!= "")

11.2 特征构建类

问题3：for 循环里的变量传不出来

现象：在 for 循环里构建的 DataFrame，循环外用不了原因：Scala 的 for 循环内部变量作用域在循环内解决：

方法1：在循环里把每个区间的结果保存到 Hive（用 append 模式）
方法2：用累加的方式（union），但数据量大时不推荐
本项目用的是方法1：循环内 append 到 Hive 表

问题4：join 之后列名对不上

现象：join 后找不到某些列原因：两个表有同名列，join 后列名可能变了解决：

用 Seq("col1", "col2") 方式 join，会自动去掉重复的 join 列
或者 join 前给列起别名

问题5：特征和标签怎么合并

现象：特征表没有 label 原因：特征是从 TimeFeatures 来的，label 在 AdData 里解决：用 ranks 字段 join 两个表，把 label 加进来

11.3 模型训练类

问题6：报 "column features not found"

现象：训练时报找不到 features 列原因：特征列名不叫 features 解决：

方法1：把特征列名改成 features
方法2：用 setFeaturesCol("你的列名") 指定
逻辑回归默认找 features 和 label，所以最好列名就叫这个

问题7：标签列类型不对

现象：报类型不匹配原因：label 是 Int 类型，模型需要 Double 解决：转成 double 类型

scala

col("label").cast("double")

问题8：集群模式运行报错

现象：本地模式没问题，集群模式报错 可能原因：

Hive 配置没同步 → 确保 hive-site.xml 在 resources 里
依赖包缺失 → 打包时要把依赖打进去，或者集群上有
路径不对 → 集群上的路径和本地不一样
权限问题 → HDFS 目录权限

11.4 模型加载类

问题9：模型类名搞混了

现象：加载模型时报找不到类原因：训练用的是算法类（LogisticRegression），加载要用模型类（LogisticRegressionModel）解决：

算法类（训练用）	模型类（加载用）
LogisticRegression	LogisticRegressionModel
RandomForestClassifier	RandomForestClassificationModel
LinearRegression	LinearRegressionModel

问题10：模型路径不对

现象：加载模型时报路径不存在原因：

模型保存在 HDFS 上，加载时要写完整 HDFS 路径
或者本地模式加载 HDFS 路径，配置不对解决：
完整路径：hdfs://master:8020/tipdm/data/model/Logistic
确保 Hadoop 配置正确

第十二部分：实习 / 面试高频考点

12.1 项目流程类（高频）

Q1：说一下你这个项目的整体流程？

这是一个广告流量作弊识别项目，用 Spark 做的综合实战。整体流程分为5步：
数据探索：170万条7天的广告流量数据，探索数据分布、缺失值、冗余字段
特征工程：根据三种作弊方式（脚本刷新、清cookie、ADSL拨号）构建了4个特征（N、N1、N2、N3）
数据预处理：删除高缺失字段，MinMaxScaler 标准化，7:3划分数据集
模型训练：分别用逻辑回归和随机森林训练，随机森林准确率91.1%更好
模型应用：保存模型，加载后对新数据预测
技术栈：Spark SQL、Spark MLlib、Hive

Q2：你构建了哪些特征？为什么这么构建？

构建了4个特征，对应三种常见的流量作弊方式：
N：同一 ip + cookie 的访问次数 → 对应脚本刷新作弊
N1：同一 ip 对应的不同 cookie 数量 → 对应定期清除 cookie 作弊
N2：ip 前两段相同的记录数 → 对应 ADSL 拨号作弊（大范围）
N3：ip 前三段相同的记录数 → 对应 ADSL 拨号作弊（小范围）
特征是根据业务理解和数据探索得出的，每种作弊方式都有其独特的数据模式。

Q3：为什么用5小时作为时间窗口？

这是根据广告点击周期的频率来划分的。作弊行为往往是集中在一段时间内的，用时间窗口来计算特征能更好地捕捉短期作弊行为。窗口太大可能漏掉短期作弊，太小可能噪声太多。

Q4：数据量多大？用什么模式运行？

原始数据约170万条，22个字段。探索和特征构建阶段用本地模式（演示用，只取前25小时），模型训练用独立集群模式（spark://master:7077），通过 spark-submit 提交。实际生产环境应该全流程都用集群。

12.2 特征工程类（高频）

Q5：缺失值怎么处理的？

先统计每个字段的缺失率：
缺失率特别高的（比如 creativeid 98%），直接删除
字符类型的缺失，无法插补，也删除
有业务含义且缺失率不高的，可以考虑填充
本项目删除了7个缺失率过高的字段。另外 creativeid 比较特殊，值为0也视为缺失。

Q6：为什么用 MinMaxScaler 而不是 StandardScaler？

两种都是特征缩放方法：
StandardScaler：标准化到均值0、标准差1，适合近似正态分布的数据
MinMaxScaler：缩放到[0,1]区间，适合需要固定范围的场景
本项目的特征（访问次数等）不服从正态分布，而且希望特征都在同一范围内，所以选择了 MinMaxScaler。

Q7：特征工程中你觉得最关键的是什么？

我觉得最关键的是业务理解。特征不是凭空想出来的，而是基于对业务的深入理解。比如这个项目，如果你不知道有哪些作弊方式，就不知道该构建什么特征。数据探索也是为了验证业务假设，发现数据中的模式。好的特征决定了模型的上限，算法只是逼近这个上限。

12.3 模型类（高频）

Q8：逻辑回归和随机森林你选哪个？为什么？

我两个都试了，对比下来随机森林更好：
逻辑回归准确率 85.8%
随机森林准确率 91.1%，提升了约5%
原因：
逻辑回归是线性模型，只能学习线性关系
随机森林是多棵决策树的集成，能学习非线性关系，泛化能力更强
但逻辑回归也有优势：训练快、可解释性强、可以作为 baseline。实际工作中一般先跑逻辑回归做基准，再试更复杂的模型。

Q9：随机森林的参数你调了哪些？

主要参数：
setNumTrees(5)：树的数量，这里设了5棵（教学演示用，实际可以更多）
setFeaturesCol("features")：特征列
setLabelCol("label")：标签列
实际工作中还会调：
setMaxDepth：树的最大深度（防止过拟合）
setMaxBins：分箱数
setMinInstancesPerNode：叶子节点最小样本数
调参一般用网格搜索或随机搜索，结合交叉验证。

Q10：怎么评估模型？用了什么指标？

用 MulticlassClassificationEvaluator 评估，指标是准确率（accuracy）。
分类问题常用指标：
准确率（accuracy）：整体预测正确的比例
精确率（precision）：预测为正的里面真正是正的比例
召回率（recall）：真正是正的里面预测对的比例
F1值：精确率和召回率的调和平均
本项目用准确率是因为比较直观，而且类别分布不是特别不平衡。如果类别很不平衡，应该看精确率、召回率或AUC。

Q11：模型过拟合了怎么办？

过拟合就是训练集上表现好，测试集上表现差。解决方法：
增加数据：更多的训练数据
正则化：逻辑回归加 regParam，随机森林减小树深度
简化模型：减少树的数量、减小树深度
特征选择：去掉不重要的特征
交叉验证：更准确地评估模型
集成学习：比如随机森林本身就是为了降低过拟合

12.4 工程类（中频）

Q12：模型怎么保存和加载？

训练好的模型用 .write.overwrite().save(path) 保存，加载用 ModelClass.load(path)。
注意点：
保存路径可以是本地路径，也可以是 HDFS 路径
加载时的类名是模型类（如 LogisticRegressionModel），不是算法类
模型保存后，加载出来效果应该和保存前一致
本项目模型保存在 HDFS 的 /tipdm/data/model/ 目录下。

Q13：为什么数据要保存到 Hive？

原因：
数据共享：不同的程序（探索、特征、训练）都要用同一份数据
数据持久化：处理好的数据保存下来，不用每次重新处理
方便查询：可以用 SQL 直接查
管理方便：有数据库、表的概念，结构清晰
本项目建了 ad_traffic 数据库，里面有 AdData、TimeFeatures、FeaturesData、ModelData、LoadData 等表。

Q14：spark-submit 怎么提交？参数怎么传？

提交命令示例：
bash
spark-submit --master spark://master:7077 \
  --class Logistic /opt/data/model.jar \
  ad_traffic.ModelData \
  /tipdm/data/model/Logistic
参数说明：
--master：集群地址
--class：主类名
JAR包路径
后面的是 main 方法的 args 参数，在代码里用 args(0)、args(1) 获取
这样做的好处是参数灵活，不用每次改代码重新打包。

12.5 综合类（高频）

Q15：这个项目你遇到的最大难点是什么？怎么解决的？

我觉得最大的难点是特征构建。
难点1：怎么从原始数据中提取出有用的特征？
解决：先做业务分析，了解常见的作弊方式，再用数据探索验证
难点2：按时间窗口构建特征时，for 循环里的结果怎么传出来？
解决：Scala 的 for 循环变量作用域在循环内，所以用了在循环内 append 到 Hive 表的方式，每个区间的结果都追加到同一张表
难点3：特征和标签不在同一张表
解决：通过 ranks 字段 join，把 label 加进来

Q16：如果让你优化这个项目，你会怎么做？

可以从几个方面优化：
特征优化：
增加更多特征（比如时间特征、设备特征）
特征选择，去掉不重要的特征
尝试特征交叉
模型优化：
调参：网格搜索找最优参数
尝试更多算法：GBDT、XGBoost、LightGBM
模型融合
工程优化：
全流程跑集群，处理全量数据
特征工程 Pipeline 化
实时预测流（Spark Streaming / Structured Streaming）
评估优化：
看更多指标：精确率、召回率、AUC
交叉验证
混淆矩阵分析

Q17：Spark 在这个项目中发挥了什么作用？

Spark 在这个项目中贯穿始终：
数据处理：Spark SQL / DataFrame 做数据探索、清洗、特征构建
机器学习：Spark MLlib 做特征工程（MinMaxScaler、VectorAssembler）和模型训练
数据存储：和 Hive 集成，方便数据管理和共享
分布式计算：数据量大的时候，集群模式分布式处理
整个项目体现了 Spark 一站式大数据处理平台的特点，从数据处理到机器学习都能在同一个框架内完成。

笔记版本：V1.0 对应教材：《Spark大数据技术与应用（第3版）》人民邮电出版社 对应项目：项目9 广告检测的流量作弊识别——Spark综合实战 最后更新：2026年8月

项目9：广告检测的流量作弊识别——Spark 综合实战 ​

目录 ​

第一部分：项目背景与需求分析 ​

1.1 项目背景 ​

什么是虚假流量？ ​

为什么会有虚假流量？ ​

本项目要做什么？ ​

1.2 需求分析 ​

实现流程 ​

三大核心任务 ​

第二部分：常见流量作弊方式 ​

2.1 三种常见作弊方式 ​

方式一：脚本刷新网页作弊 ​

方式二：定期清除 cookie，刷新网页作弊 ​

方式三：ADSL 重新拨号后刷新网页作弊 ​

三种作弊方式对比 ​

第三部分：数据说明 ​

3.1 数据文件 ​

3.2 主要字段 ​

第四部分：任务9.1-9.2 数据探索分析 ​

4.1 任务目标 ​

4.2 基础探索 ​

步骤1：创建 SparkSession ​

步骤2：读取数据并统计记录数 ​

步骤3：统计日流量 ​

步骤4：分析数据类型 ​

步骤5：统计缺失数据 ​

自定义缺失率计算函数 ​

调用函数统计所有字段 ​

步骤6：冗余数据分析 ​

4.3 业务探索：作弊特征分析 ​

探索一：脚本刷新网页作弊 ​

探索二：定期清除 cookie 作弊 ​

探索三：ADSL 重新拨号作弊 ​

按 ip 前两段统计 ​

按 ip 前三段统计 ​

4.4 探索分析小结 ​

第五部分：任务9.3 数据预处理与特征构建 ​

5.1 数据预处理 ​

删除缺失率过高的字段 ​

保存到 Hive ​

准备工作 ​

保存数据 ​

5.2 特征构建 ​

构建哪几个特征？ ​

为什么按时间区间构建特征？ ​

时间区间划分 ​

循环构建特征 ​

代码说明 ​

合并特征和标签 ​

第六部分：特征标准化与数据划分 ​

6.1 为什么要标准化？ ​

6.2 使用 MinMaxScaler 归一化 ​

数据划分说明 ​

第七部分：任务9.4 逻辑回归模型构建与评估 ​

7.1 为什么用逻辑回归？ ​

7.2 代码实现 ​

代码说明 ​

7.3 提交到集群运行 ​

打包 JAR ​

上传到 Linux ​

spark-submit 提交 ​

参数说明 ​

7.4 模型效果 ​

第八部分：随机森林模型构建与评估 ​

8.1 为什么用随机森林？ ​

8.2 代码实现 ​

8.3 提交到集群运行 ​

8.4 模型效果对比 ​

为什么随机森林更好？ ​

第九部分：模型加载与应用 ​

9.1 为什么要加载模型？ ​

9.2 代码实现 ​

代码说明 ​

9.3 结果说明 ​

第十部分：完整项目流程总结 ​

10.1 整体流程图 ​

10.2 各阶段使用的运行模式 ​

10.3 用到的 Spark 技术 ​

第十一部分：常见问题与排错指南 ​

项目9：广告检测的流量作弊识别——Spark 综合实战

目录

第一部分：项目背景与需求分析

1.1 项目背景

什么是虚假流量？

为什么会有虚假流量？

本项目要做什么？

1.2 需求分析

实现流程

三大核心任务

第二部分：常见流量作弊方式

2.1 三种常见作弊方式

方式一：脚本刷新网页作弊

方式二：定期清除 cookie，刷新网页作弊

方式三：ADSL 重新拨号后刷新网页作弊

三种作弊方式对比

第三部分：数据说明

3.1 数据文件

3.2 主要字段

第四部分：任务9.1-9.2 数据探索分析

4.1 任务目标

4.2 基础探索

步骤1：创建 SparkSession

步骤2：读取数据并统计记录数

步骤3：统计日流量

步骤4：分析数据类型

步骤5：统计缺失数据

自定义缺失率计算函数

调用函数统计所有字段

步骤6：冗余数据分析

4.3 业务探索：作弊特征分析

探索一：脚本刷新网页作弊

探索二：定期清除 cookie 作弊

探索三：ADSL 重新拨号作弊

按 ip 前两段统计

按 ip 前三段统计

4.4 探索分析小结

第五部分：任务9.3 数据预处理与特征构建

5.1 数据预处理

删除缺失率过高的字段

保存到 Hive

准备工作

保存数据

5.2 特征构建

构建哪几个特征？

为什么按时间区间构建特征？

时间区间划分

循环构建特征

代码说明

合并特征和标签

第六部分：特征标准化与数据划分

6.1 为什么要标准化？

6.2 使用 MinMaxScaler 归一化

数据划分说明

第七部分：任务9.4 逻辑回归模型构建与评估

7.1 为什么用逻辑回归？

7.2 代码实现

代码说明

7.3 提交到集群运行

打包 JAR

上传到 Linux

spark-submit 提交

参数说明

7.4 模型效果

第八部分：随机森林模型构建与评估

8.1 为什么用随机森林？

8.2 代码实现

8.3 提交到集群运行

8.4 模型效果对比

为什么随机森林更好？

第九部分：模型加载与应用

9.1 为什么要加载模型？

9.2 代码实现

代码说明

9.3 结果说明

第十部分：完整项目流程总结

10.1 整体流程图

10.2 各阶段使用的运行模式

10.3 用到的 Spark 技术

第十一部分：常见问题与排错指南