项目8：饮用水源合格性预测——Spark MLlib 机器学习算法库

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

第一部分：项目背景与机器学习概述
第二部分：机器学习基本概念
第三部分：Spark MLlib 简介与发展
第四部分：MLlib 数据类型
第五部分：特征提取
第六部分：数据特征处理（标准化/归一化）
第七部分：回归算法
第八部分：分类算法
第九部分：聚类算法（K-Means）
第十部分：关联规则（FP-Growth）
第十一部分：推荐算法（ALS）
第十二部分：模型评估
第十三部分：项目实战——饮用水源合格性预测
第十四部分：常见问题与排错指南
第十五部分：实习 / 面试高频考点
附录：习题解析

第一部分：项目背景与机器学习概述

1.1 项目背景

为什么需要机器学习预测？

水是生命之源，饮用水安全直接关系人类健康。随着工业化、城市化加快，水源污染问题日益严重。传统的人工检测方式效率低、成本高。

机器学习能做什么？

用历史数据训练模型
自动预测水源是否合格
快速、准确、可扩展

项目场景

现有一份饮用水源指标检测数据（约8000条模拟数据），包含20个水质指标和1个合格性标签。

本项目要做的事：

读取并探索水质数据
数据清洗与标准化
用随机森林算法构建预测模型
评估模型准确率

第二部分：机器学习基本概念

2.1 什么是机器学习？

机器学习就是让机器像人一样具备学习能力，从数据中自动发现规律，并用规律来预测或决策。

机器学习的过程

历史数据 → 训练模型 → 评估模型 → 达标？→ 是 → 用于预测
                              ↓ 否
                           调整算法 → 重新训练

💡 形象比喻：就像教小孩认猫：
给小孩看很多猫的照片（训练数据）
小孩总结出"猫有尖耳朵、长胡须、圆眼睛"（模型）
拿新照片考小孩（测试）
认对了就过关，认错了就再教（调整）

2.2 机器学习的分类

按训练数据是否有标签分

类型	特点	典型算法	应用场景
监督学习	训练数据有标签（已知答案）	线性回归、逻辑回归、决策树、朴素贝叶斯、随机森林	预测、分类
无监督学习	训练数据没有标签（不知道答案）	K-Means聚类、FP-Growth关联规则	聚类、关联分析、异常检测
半监督学习	大部分没标签，小部分有标签	-	数据量大但标注成本高的场景

监督学习 vs 无监督学习

对比项	监督学习	无监督学习
训练数据	有标签	无标签
目标	预测/分类	发现结构/模式
评估	有明确对错	没有标准答案
例子	预测房价、判断垃圾邮件	用户分群、商品关联推荐

💡 简单理解：
监督学习 = "有标准答案的练习题"
无监督学习 = "自己找规律的探索题"

2.3 常见机器学习算法类型

算法类型	作用	代表算法
回归算法	预测连续值（比如房价、温度）	线性回归、逻辑回归
分类算法	判断类别（比如是否合格、是猫是狗）	朴素贝叶斯、决策树、随机森林
聚类算法	自动分组（比如用户分群）	K-Means
推荐算法	给用户推荐物品	协同过滤（ALS）
降维算法	减少特征维度	PCA
关联规则	发现物品之间的关联	FP-Growth

第三部分：Spark MLlib 简介与发展

3.1 什么是 MLlib？

MLlib 是 Spark 的机器学习库，是 Spark 四大组件之一。

基于 Spark 的分布式计算能力
可以处理海量数据的机器学习任务
提供了丰富的算法和工具

MLlib 包含的内容

模块	说明
算法	分类、回归、聚类、协同过滤等
特征化	特征提取、转换、降维、选择
管道（Pipeline）	构建机器学习工作流的工具
持久化	保存和加载模型、管道
实用工具	线性代数、统计、数据处理

3.2 MLlib 的发展历史

版本	重要变化
0.8	MLlib 加入 Spark，只支持 Java 和 Scala
1.0	开始支持 Python
1.2	分为两个包：spark.mllib（基于RDD）和 spark.ml（基于DataFrame）
2.0+	基于 RDD 的 API 进入维护模式，官方推荐使用 spark.ml 包

两个包的区别

包名	底层	特点	推荐度
`spark.mllib`	RDD	老版本API，基于RDD	⚠️ 维护模式，不再新增
`spark.ml`	DataFrame	新版本API，支持Pipeline工作流	✅ 官方推荐

💡 记忆：
mllib（多一个l）= 老的，基于RDD
ml（少一个l）= 新的，基于DataFrame，推荐用这个
就像 Spark SQL 比 RDD 更高级一样，spark.ml 比 spark.mllib 更高级

第四部分：MLlib 数据类型

4.1 常见数据类型

MLlib 中常用的基本数据类型：

类型	说明	适用场景
向量（Vector）	一组数值，可以是稠密或稀疏的	特征向量
标注点（LabeledPoint）	标签 + 特征向量	监督学习
矩阵（Matrix）	二维数值数组	数学计算
Rating	用户ID + 物品ID + 评分	推荐算法

⚠️ 注意：RDD 是 Spark 的核心数据结构，不是 MLlib 特有的数据类型。

4.2 向量（Vector）

向量是 MLlib 中最基础的数据类型，表示一组特征值。

两种向量

类型	说明	适用场景
稠密向量（Dense）	所有值都存	大部分值不为0
稀疏向量（Sparse）	只存非零值的索引和值	大部分值为0（节省空间）

示例

scala

import org.apache.spark.ml.linalg.Vectors

// 稠密向量：[1.0, 0.0, 0.0, 3.0]
val denseVec = Vectors.dense(1.0, 0.0, 0.0, 3.0)

// 稀疏向量：4维，索引0和3有值
val sparseVec = Vectors.sparse(4, Seq((0, 1.0), (3, 3.0)))

4.3 标注点（LabeledPoint）

标注点 = 标签（label） + 特征向量（features）

用于监督学习
标签是已知的答案

scala

import org.apache.spark.ml.feature.LabeledPoint
import org.apache.spark.ml.linalg.Vectors

// 标签为1.0，特征向量是[0.0, 1.0, 2.0]
val pos = LabeledPoint(1.0, Vectors.dense(0.0, 1.0, 2.0))

💡 理解： LabeledPoint 就像"带答案的练习题"， label 是答案，features 是题目。

第五部分：特征提取

特征提取就是把原始数据转换成机器学习算法能用的特征向量。

5.1 TF-IDF 算法

什么是 TF-IDF？

TF-IDF 是一种文本特征提取方法，用来衡量一个词在文档中的重要程度。

两个部分

缩写	全称	含义
TF	Term Frequency（词频）	一个词在文档中出现的频率
IDF	Inverse Document Frequency（逆文档频率）	一个词在整个语料库中是否常见

公式

TF = 词在文档中出现的次数 / 文档总词数
IDF = log(总文档数 / (包含该词的文档数 + 1))
TF-IDF = TF × IDF

💡 理解：
TF高 = 这个词在本文档中出现多 → 可能重要
IDF高 = 这个词在其他文档中很少出现 → 更有区分度
TF-IDF高 = 这个词对本文档很重要，而且在其他地方不常见

在 Spark MLlib 中的实现

HashingTF：计算词频向量（用哈希法）
IDF：计算逆文档频率，调整特征向量

spark.mllib 包实现（基于RDD）

scala

import org.apache.spark.mllib.feature.{HashingTF, IDF}

// 1. 读取文档，每行一个文档，按空格分词
val documents = sc.textFile("/tipdm/data/tf-idf.txt")
  .map(_.split(" ").toSeq)

// 2. 计算TF（词频）
val hashingTF = new HashingTF()
val tf = hashingTF.transform(documents)
tf.cache()

// 3. 计算IDF并转换
val idf = new IDF().fit(tf)
val tfidf = idf.transform(tf)

// 4. 查看结果
tfidf.collect().foreach(println)

spark.ml 包实现（基于DataFrame，推荐）

scala

import org.apache.spark.ml.feature.{HashingTF, IDF, Tokenizer}

// 1. 准备数据
val sentenceData = spark.createDataFrame(Seq(
  (0.0, "Hi I heard about Spark"),
  (0.0, "I wish Java could use case classes"),
  (1.0, "Logistic regression models are neat")
)).toDF("label", "sentence")

// 2. 分词（句子 → 单词列表）
val tokenizer = new Tokenizer()
  .setInputCol("sentence")
  .setOutputCol("words")
val wordsData = tokenizer.transform(sentenceData)

// 3. 计算TF
val hashingTF = new HashingTF()
  .setInputCol("words")
  .setOutputCol("rawFeatures")
  .setNumFeatures(20)
val featurizedData = hashingTF.transform(wordsData)

// 4. 计算IDF
val idf = new IDF()
  .setInputCol("rawFeatures")
  .setOutputCol("features")
val idfModel = idf.fit(featurizedData)
val rescaledData = idfModel.transform(featurizedData)

// 5. 查看结果
rescaledData.select("features").show(false)

⚠️ 哈希冲突问题： HashingTF 用哈希函数把词映射到索引，可能不同的词映射到同一个位置（哈希冲突）。解决方法：调大 setNumFeatures() 的值（特征维度）。

5.2 Word2Vec 算法

什么是 Word2Vec？

Word2Vec 是自然语言处理（NLP）中的重要算法，把词转换成向量。

向量能表示词的语义信息
语义相近的词，向量也相近
可以用来找同义词、做文本分类等

spark.mllib 包实现

scala

import org.apache.spark.mllib.feature.{Word2Vec, Word2VecModel}

// 1. 读取语料库，每行一个文档
val input = sc.textFile("/tipdm/data/w2v.txt")
  .map(line => line.split(" ").toSeq)

// 2. 训练Word2Vec模型
val word2vec = new Word2Vec().setMinCount(0)
val model = word2vec.fit(input)

// 3. 找与"spark"最相似的5个词
val synonyms = model.findSynonyms("spark", 5)
for ((synonym, cosineSimilarity) <- synonyms) {
  println(s"$synonym $cosineSimilarity")
}

spark.ml 包实现

scala

import org.apache.spark.ml.feature.Word2Vec

// 1. 准备数据
val documentDF = spark.createDataFrame(Seq(
  "spark is a fast and general engine for large scale data processing".split(" "),
  "word2vec is a popular algorithm for learning word embeddings".split(" "),
  "machine learning algorithms require large amounts of training data".split(" "),
  "natural language processing involves understanding human language".split(" ")
).map(Tuple1.apply)).toDF("text")

// 2. 训练模型
val word2Vec = new Word2Vec()
  .setInputCol("text")
  .setOutputCol("result")
  .setMinCount(0)
val model = word2Vec.fit(documentDF)

// 3. 找相似词
model.findSynonyms("spark", 5).show()

5.3 统计方法

spark.mllib 包（基于RDD）

scala

import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.stat.Statistics

// 1. 读取数据
val data = sc.textFile("/tipdm/data/stat.txt")
  .map(_.split(" "))
  .map(f => f.map(_.toDouble))
val data1 = data.map(f => Vectors.dense(f))

// 2. 列统计（均值、方差等）
val stat1 = Statistics.colStats(data1)
stat1.mean       // 均值
stat1.variance   // 方差

// 3. 相关系数矩阵
val corr1 = Statistics.corr(data1, "pearson")

spark.ml 包（基于DataFrame）

scala

import org.apache.spark.ml.linalg.{Matrix, Vectors}
import org.apache.spark.ml.stat.{Correlation, ChiSquareTest}
import org.apache.spark.sql.Row

// 1. 准备数据
val data = Seq(
  (0.0, Vectors.dense(0.5, 10.0)),
  (0.0, Vectors.dense(1.5, 20.0)),
  (1.0, Vectors.dense(1.5, 30.0)),
  (0.0, Vectors.dense(3.5, 30.0)),
  (0.0, Vectors.dense(3.5, 40.0)),
  (1.0, Vectors.dense(3.5, 40.0))
)
val dataFrame = data.toDF("label", "features")

// 2. 相关系数矩阵
val Row(coeff1: Matrix) = Correlation.corr(dataFrame, "features").head()

// 3. 卡方独立性检验
val chi = ChiSquareTest.test(dataFrame, "features", "label").head()

第六部分：数据特征处理（标准化/归一化）

6.1 为什么要做特征处理？

不同特征的量纲和量级不同，会影响模型效果。比如：

年龄：0-100
收入：0-100000

收入的数值大，会在计算中"压倒"年龄的影响。

标准化/归一化就是把不同特征放到同一个尺度上，让每个特征的贡献更公平。

6.2 三种常用方法对比

方法	作用对象	效果	常用参数
Normalizer	每一行（每个样本）	行向量的范数=1	`setP`（p范数，默认2）
StandardScaler	每一列（每个特征）	均值为0，标准差为1（标准化）	`withStd`、`withMean`
MinMaxScaler	每一列（每个特征）	缩放到[0,1]区间（归一化）	`min`、`max`

💡 记忆：
Normalizer = 按行归一化（每个样本单独处理）
StandardScaler / MinMaxScaler = 按列处理（每个特征单独处理）

准备数据

scala

import org.apache.spark.ml.linalg.Vectors

val dataFrame = spark.createDataFrame(Seq(
  (0, Vectors.dense(1.0, 0.5, -1.0)),
  (1, Vectors.dense(2.0, 1.0, 1.0)),
  (2, Vectors.dense(4.0, 10.0, 2.0))
)).toDF("id", "features")

6.3 Normalizer（行归一化）

作用

把每一行向量转换成单位范数（长度为1）。

示例

scala

import org.apache.spark.ml.feature.Normalizer

val normalizer = new Normalizer()
  .setInputCol("features")
  .setOutputCol("normFeatures")
  .setP(1.0)  // L1范数

val l1NormData = normalizer.transform(dataFrame)
l1NormData.show()

参数

参数	说明	默认值
`setP`	p-norm的p值	2（L2范数）

💡 理解： L1范数 = 所有元素绝对值之和 = 1 L2范数 = 所有元素平方和开根号 = 1

6.4 StandardScaler（标准化）

作用

把每一列特征标准化为：均值为0，标准差为1（Z-score标准化）。

公式

标准化后的值 = (原值 - 均值) / 标准差

示例

scala

import org.apache.spark.ml.feature.StandardScaler

val scaler = new StandardScaler()
  .setInputCol("features")
  .setOutputCol("scaledFeatures")
  .setWithStd(true)    // 标准化到单位标准差
  .setWithMean(false)  // 是否变换为0均值

val scalerModel = scaler.fit(dataFrame)
val scaledData = scalerModel.transform(dataFrame)
scaledData.show(false)

参数

参数	说明	默认值
`withStd`	是否缩放到单位标准差	true
`withMean`	是否减去均值（中心化）	false

⚠️ 注意： withMean=true 会返回稠密输出，不适合稀疏数据。 WithMean() 只是 StandardScaler 的一个参数，不是独立的方法！

6.5 MinMaxScaler（最小-最大归一化）

作用

把每一列特征线性映射到指定区间（默认[0,1]）。

公式

归一化后的值 = (原值 - 最小值) / (最大值 - 最小值)

示例

scala

import org.apache.spark.ml.feature.MinMaxScaler

val scaler = new MinMaxScaler()
  .setInputCol("features")
  .setOutputCol("scaledFeatures")

val scalerModel = scaler.fit(dataFrame)
val scaledData = scalerModel.transform(dataFrame)
scaledData.show(false)

参数

参数	说明	默认值
`min`	区间下限	0
`max`	区间上限	1

6.6 三种方法总结

方法	处理维度	结果范围	适用场景
Normalizer	行（样本）	范数=1	需要统一样本尺度
StandardScaler	列（特征）	均值0，标准差1	数据近似正态分布
MinMaxScaler	列（特征）	[0,1]（默认）	需要固定范围

第七部分：回归算法

7.1 线性回归

什么是线性回归？

用一组特征的线性组合来预测连续值。

公式

预测值 = w1×x1 + w2×x2 + ... + wn×xn + b

x1, x2...xn 是特征
w1, w2...wn 是权重（模型要学的参数）
b 是偏置

数据格式：LIBSVM 格式

标签 索引1:值1 索引2:值2 索引3:值3 ...

第一列：标签（预测值）
后面：特征列，格式为"索引:值"
索引从1开始

示例代码

scala

import org.apache.spark.ml.regression.{LinearRegression, LinearRegressionModel}

// 1. 读取LIBSVM格式数据
val data = spark.read.format("libsvm")
  .load("/tipdm/data/sample_linear_regression_data.txt")

// 2. 划分训练集和测试集（7:3）
val Array(train, test) = data.randomSplit(Array(0.7, 0.3))

// 3. 创建线性回归模型
val lr = new LinearRegression()
  .setMaxIter(10)              // 最大迭代次数
  .setRegParam(0.3)            // 正则化参数
  .setElasticNetParam(0.8)     // 弹性网络混合参数

// 4. 训练模型
val lrModel = lr.fit(train)

// 5. 预测
val pre = lrModel.transform(test)

// 6. 保存和加载模型
lrModel.save("myModelPath")
val model = LinearRegressionModel.load("myModelPath")

主要参数

参数	说明
`setMaxIter`	最大迭代次数
`setRegParam`	正则化参数（防止过拟合）
`setElasticNetParam`	弹性网络混合参数（L1和L2的混合）

7.2 逻辑回归

什么是逻辑回归？

虽然叫"回归"，但实际上是二分类算法。

预测的是概率（0到1之间）
概率大于阈值 → 类别A，小于阈值 → 类别B

与线性回归的区别

对比项	线性回归	逻辑回归
输出	连续值	概率（0-1）
用途	预测数值	二分类
例子	预测房价	判断是否合格

包位置

spark.ml.classification.LogisticRegression

💡 注意：逻辑回归在 classification 包，不在 regression 包！因为它本质是分类算法。

第八部分：分类算法

8.1 朴素贝叶斯

什么是朴素贝叶斯？

基于贝叶斯定理的分类算法，"朴素"是指假设特征之间相互独立。

简单、快速
适合文本分类、垃圾邮件识别等

原理

对于给定的特征，计算属于每个类别的概率，选概率最大的那个类别。

示例：根据天气判断是否打球

scala

import org.apache.spark.ml.classification.NaiveBayes
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator

// 1. 读取数据（LIBSVM格式）
// 第1列：是否打球（标签）
// 后面：天气、温度、湿度、是否刮风（特征）
val data = spark.read.format("libsvm")
  .load("/tipdm/data/weather_data.txt")

// 2. 划分训练集和测试集（6:4）
val Array(trainingData, testData) = data.randomSplit(Array(0.6, 0.4))

// 3. 训练朴素贝叶斯模型
val model = new NaiveBayes().fit(trainingData)

// 4. 预测
val predictions = model.transform(testData)
predictions.show()

// 5. 评估准确率
val evaluator = new MulticlassClassificationEvaluator()
  .setLabelCol("label")
  .setPredictionCol("prediction")
  .setMetricName("accuracy")

val accuracy = evaluator.evaluate(predictions)
println(s"Test set accuracy = $accuracy")

⚠️ 注意：朴素贝叶斯要求输入是数值类型，如果原始数据是字符类型（如"晴""雨"），需要先转换成数值。

8.2 决策树

什么是决策树？

像一棵树一样，从根节点开始，根据特征值一步步往下走，最后到达叶子节点得到结果。

特点

容易理解和解释
可以处理类别特征
适合多分类问题
容易过拟合（太深的树）

包位置

spark.ml.classification.DecisionTreeClassifier

原理

每个节点选择一个特征进行分裂，选择的原则是让分裂后的结果"最纯"（信息增益最大）。

💡 形象比喻：决策树就像"二十问"游戏，每问一个问题（特征），就把范围缩小，问到最后就知道答案了。

第九部分：聚类算法（K-Means）

9.1 什么是聚类？

聚类是无监督学习，数据没有标签，算法自动把相似的数据分到同一类。

聚类 vs 分类

对比项	分类	聚类
学习方式	监督学习	无监督学习
训练数据	有标签	无标签
目标	已知类别，判断新数据属于哪类	不知道类别，自动分组
例子	判断邮件是否垃圾	用户分群

9.2 K-Means 算法

什么是 K-Means？

最常用的聚类算法，把数据分成 K 个簇。

核心思想

随机选 K 个点作为初始中心点
把每个数据点分配到最近的中心点
重新计算每个簇的中心点
重复2-3，直到中心点不再变化

K 的含义

K = 要分成几类，需要人为指定。

示例代码

scala

import org.apache.spark.ml.clustering.KMeans
import org.apache.spark.ml.evaluation.ClusteringEvaluator

// 1. 读取数据
val dataset = spark.read.format("libsvm")
  .load("/tipdm/data/sample_kmeans_data.txt")

// 2. 创建K-Means模型，K=2
val kmeans = new KMeans().setK(2).setSeed(1L)

// 3. 训练
val model = kmeans.fit(dataset)

// 4. 预测（给每个数据点分配簇）
val predictions = model.transform(dataset)

// 5. 评估（轮廓系数）
val evaluator = new ClusteringEvaluator()
val silhouette = evaluator.evaluate(predictions)
println(s"轮廓系数：$silhouette")

// 6. 查看聚类中心
model.clusterCenters.foreach(println)
predictions.show(false)

主要参数

参数	说明
`setK`	簇的数量（K值）
`setSeed`	随机种子（保证结果可复现）

评估指标：轮廓系数（silhouette）

取值范围：[-1, 1]
越接近1，聚类效果越好
越接近-1，聚类效果越差
接近0，说明在边界上

💡 理解轮廓系数：一个点和自己簇里的点越近、和其他簇的点越远，轮廓系数就越高。

第十部分：关联规则（FP-Growth）

10.1 什么是关联规则？

关联规则挖掘就是发现数据中"经常一起出现"的模式。经典例子："买了啤酒的人经常买尿布"。

两个核心概念

概念	含义	说明
支持度（Support）	某组物品一起出现的频率	出现次数 / 总订单数
置信度（Confidence）	买了A的人中，同时买了B的比例	P(B\|A) = P(AB) / P(A)

强关联规则

同时满足最小支持度和最小置信度的规则。

10.2 FP-Growth 算法

FP-Growth 是高效的频繁项集挖掘算法。

FP = Frequent Pattern（频繁模式）
比传统的 Apriori 算法更快

示例：餐饮点餐数据

scala

import org.apache.spark.ml.fpm.FPGrowth

// 1. 读取数据
val rawDF = spark.read.text("/tipdm/data/menu_orders.txt")

// 2. 处理数据：每行一个订单，菜品用空格分隔
val transactionsDF = rawDF.withColumn("items",
  split(trim(col("value")), " "))

// 3. 创建FP-Growth模型
val fpgrowth = new FPGrowth()
  .setItemsCol("items")
  .setMinSupport(0.5)      // 最小支持度
  .setMinConfidence(0.6)   // 最小置信度

// 4. 训练
val model = fpgrowth.fit(transactionsDF)

// 5. 查看频繁项集
model.freqItemsets.show()

// 6. 查看关联规则
model.associationRules.show()

// 7. 用规则做预测/推荐
model.transform(transactionsDF).show()

主要参数

参数	说明
`setMinSupport`	最小支持度（0-1）
`setMinConfidence`	最小置信度（0-1）

💡 应用场景：
超市购物篮分析（啤酒与尿布）
推荐系统（买了这个还买了那个）
网页点击路径分析

第十一部分：推荐算法（ALS）

11.1 什么是协同过滤？

协同过滤是推荐系统常用算法，基于"相似用户喜欢相似物品"的思想。

两种思路

类型	思路	例子
基于用户	和你相似的人喜欢的，你也可能喜欢	和你口味相似的人都看了这部电影，推荐给你
基于物品	和你喜欢的物品相似的，你也可能喜欢	你喜欢这部电影，推荐类似的电影

11.2 ALS 算法

ALS（Alternating Least Squares，交替最小二乘法）是 Spark MLlib 实现的协同过滤算法。

两种模式

模式	说明
显式评分	用户有明确评分（1-5星），预测的也是评分
隐式反馈	只有交互行为（点击/浏览），预测的是置信度

数据格式

用户ID::物品ID::评分::时间戳

主要参数

参数	说明	默认值
`rank`	特征向量的大小	10
`maxIter`	最大迭代次数	10
`regParam`	正则化参数	1.0
`alpha`	隐式反馈的置信度参数	1.0

示例代码

scala

import org.apache.spark.ml.recommendation.ALS

// 1. 定义数据结构
case class Rating(userId: Int, movieId: Int, rating: Float, timestamp: Long)

def parseRating(str: String): Rating = {
  val fields = str.split("::")
  assert(fields.size == 4)
  Rating(fields(0).toInt, fields(1).toInt, fields(2).toFloat, fields(3).toLong)
}

// 2. 读取数据
val ratings = spark.read.textFile("/tipdm/data/sample_movielens_ratings.txt")
  .map(parseRating).toDF()

// 3. 划分训练集和测试集（8:2）
val Array(training, test) = ratings.randomSplit(Array(0.8, 0.2))

// 4. 构建ALS模型
val als = new ALS()
  .setMaxIter(5)
  .setRegParam(0.01)
  .setUserCol("userId")
  .setItemCol("movieId")
  .setRatingCol("rating")

// 5. 训练
val model = als.fit(training)

// 6. 设置冷启动策略（遇到新用户/新物品时丢弃）
model.setColdStartStrategy("drop")

// 7. 预测
val predictions = model.transform(test)

// 8. 为所有用户推荐Top10电影
val userRecs = model.recommendForAllUsers(10)

// 9. 为所有电影推荐Top10用户
val movieRecs = model.recommendForAllItems(10)

// 10. 为指定用户推荐
val users = ratings.select(als.getUserCol).distinct().limit(3)
val userSubsetRecs = model.recommendForUserSubset(users, 10)

// 11. 为指定电影推荐
val movies = ratings.select(als.getItemCol).distinct().limit(3)
val movieSubSetRecs = model.recommendForItemSubset(movies, 10)

冷启动问题

问题：测试集中有训练集没见过的新用户/新物品，没法预测
解决：setColdStartStrategy("drop")，遇到这种情况直接丢弃

💡 理解 ALS： ALS 把用户-物品评分矩阵分解成两个小矩阵：
用户特征矩阵（每个用户一个特征向量）
物品特征矩阵（每个物品一个特征向量）交替优化这两个矩阵，直到收敛。预测时，用户向量 × 物品向量 = 预测评分。

第十二部分：模型评估

12.1 为什么要评估模型？

训练完模型后，怎么知道模型好不好？

分类问题：预测准不准？
回归问题：预测误差大不大？

模型评估就是用指标来量化模型的好坏。

12.2 分类问题评估

评估器

spark.ml.evaluation.MulticlassClassificationEvaluator

常用指标

指标	含义	说明
准确率（accuracy）	预测正确的比例	正确数 / 总数
精确率（precision）	预测为正的里面，真正是正的比例	-
召回率（recall）	真正是正的里面，预测对的比例	-
F值（F1）	精确率和召回率的调和平均	综合指标

示例

scala

import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator

val evaluator = new MulticlassClassificationEvaluator()
  .setLabelCol("label")
  .setPredictionCol("prediction")
  .setMetricName("accuracy")  // 准确率

val accuracy = evaluator.evaluate(predictions)

12.3 回归问题评估

评估器

spark.ml.evaluation.RegressionEvaluator

常用指标

指标	全称	含义	越小越好？
MSE	均方误差	误差平方的平均值	✅
RMSE	均方根误差	MSE开根号	✅
MAE	平均绝对误差	误差绝对值的平均值	✅
R²	决定系数	模型解释了多少方差	❌ 越大越好

示例

scala

import org.apache.spark.ml.evaluation.RegressionEvaluator

val evaluator = new RegressionEvaluator()
  .setLabelCol("label")
  .setPredictionCol("prediction")
  .setMetricName("mae")  // 平均绝对误差

val mae = evaluator.evaluate(predictions)

12.4 评估器总结

问题类型	评估器	常用指标
分类	MulticlassClassificationEvaluator	accuracy、precision、recall、f1
回归	RegressionEvaluator	mse、rmse、mae、r2
聚类	ClusteringEvaluator	silhouette（轮廓系数）

第十三部分：项目实战——饮用水源合格性预测

13.0 数据说明

数据文件

waterQuality.csv：约8000条饮用水源检测数据

数据字段

字段	说明
前20列	各种水质指标（铝、镉、氯胺、铬、砷、银等）
`is_safe`	是否合格（标签列），0=不合格，1=合格

项目目标

用随机森林算法构建分类模型，预测饮用水源是否合格。

任务8.1：读取数据

需求

读取 HDFS 上的水质数据，创建 DataFrame。

实现代码

scala

import org.apache.spark.sql.SparkSession

object water {
  def main(args: Array[String]): Unit = {
    // 1. 创建SparkSession
    val spark = SparkSession.builder()
      .appName("water")
      .master("local[*]")
      .enableHiveSupport()
      .getOrCreate()
    
    // 2. 设置日志级别
    spark.sparkContext.setLogLevel("WARN")
    
    // 3. 读取CSV数据
    val waterData = spark.read
      .option("header", "true")
      .option("inferSchema", "true")
      .csv("hdfs://master:8020/tipdm/data/waterQuality.csv")
    
    // 4. 查看前5行
    waterData.show(5)
  }
}

代码说明

代码	说明
`.option("header", "true")`	第一行是表头
`.option("inferSchema", "true")`	自动推断字段类型

任务8.2：探索性数据分析

需求

检查重复记录和空值
统计合格/不合格的数量
计算标签列与各特征的相关系数

实现代码

scala

// 1. 探索去重前后的数据总行数
println("去重前的数据总行数：" + waterData.count())
println("去重后的数据总行数：" + waterData.distinct().count())

// 2. 探索各字段的空值数量
waterData.select(waterData.columns.map(
  colName => sum(when(col(colName).isNull, 1).otherwise(0)).alias(colName)
): _*).show()

// 3. 探索水源合格与不合格的记录数量
waterData.groupBy("is_safe").count().show()

// 4. 创建特征向量组装器，将所有列合并为一个特征向量
val assembler = new VectorAssembler()
  .setInputCols(waterData.columns)
  .setOutputCol("features")
val waterDataWithFeatures = assembler.transform(waterData)

// 5. 计算特征之间的相关性矩阵
val Row(corrMatrix: Matrix) = Correlation.corr(waterDataWithFeatures, "features").head()

// 6. 获取特征列数
val numRows = waterData.columns.length
val corrArray = corrMatrix.toArray

// 7. 遍历所有列，计算每列与最后一列（标签列）的相关系数
for (i <- 0 until numRows - 1) {
  val correlationValue = corrArray(i * numRows + (numRows - 1))
  println(f"${waterData.columns.last}列与${waterData.columns(i)}列的相关系数：$correlationValue")
}

结果说明

共7996条数据
合格（is_safe=1）：912条
不合格（is_safe=0）：7084条
不合格率高达88.6%
与标签相关性较强的特征：铝、镉、氯胺、铬、砷、银等

💡 相关系数的意义：
绝对值越大，相关性越强
正数 = 正相关（特征越大，越可能合格）
负数 = 负相关（特征越大，越可能不合格）
一般 |r| > 0.1 就算有一定相关性

任务8.3：数据标准化

需求

对特征列进行标准化处理。注意：标签列（is_safe）不参与标准化。

实现代码

scala

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

// 1. 将除标签列外的其他列合并为一个特征向量
// waterData.columns.init 表示除最后一列外的所有列
val newAssembler = new VectorAssembler()
  .setInputCols(waterData.columns.init)
  .setOutputCol("newFeatures")

// 2. 数据标准化
val scaler = new StandardScaler()
  .setInputCol("newFeatures")
  .setOutputCol("scaledFeatures")
  .setWithStd(true)    // 缩放到单位标准差
  .setWithMean(false)  // 不减去均值

// 注意：这里只定义了scaler，fit和transform在Pipeline中做

代码说明

代码	说明
`waterData.columns.init`	除最后一列（标签列）外的所有列名
`setWithStd(true)`	标准化到标准差为1
`setWithMean(false)`	不做中心化（保持稀疏性）

💡 为什么标签列不参与标准化？ 标签是答案（0或1），不需要也不应该做标准化。标准化是针对特征的，让特征在同一尺度上。

任务8.4：构建随机森林模型

什么是随机森林？

随机森林 = 很多棵决策树一起投票。

每棵树用随机的样本和随机的特征训练
预测时，所有树投票，少数服从多数
比单棵决策树更准确，不容易过拟合

实现代码

scala

import org.apache.spark.ml.classification.RandomForestClassifier
import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator

// 1. 创建随机森林分类器
val rf = new RandomForestClassifier()
  .setFeaturesCol("scaledFeatures")   // 特征列
  .setLabelCol("is_safe")             // 标签列
  .setPredictionCol("prediction")     // 预测结果列
  .setNumTrees(100)                   // 树的数量
  .setMaxDepth(10)                    // 每棵树的最大深度
  .setMaxBins(32)                     // 特征分割的最大分箱数

// 2. 划分训练集和测试集（8:2）
val Array(trainingData, testData) = waterData.randomSplit(
  Array(0.8, 0.2), seed = 12345)

// 3. 构建管道（Pipeline）：特征组装 → 标准化 → 随机森林
val pipeline = new Pipeline()
  .setStages(Array(newAssembler, scaler, rf))

// 4. 训练模型
val model = pipeline.fit(trainingData)

// 5. 模型预测
val predictions = model.transform(testData)

// 6. 评估模型
val evaluator = new MulticlassClassificationEvaluator()
  .setLabelCol("is_safe")
  .setPredictionCol("prediction")
  .setMetricName("accuracy")

// 7. 计算准确率
val accuracy = evaluator.evaluate(predictions)

// 8. 显示结果
predictions.select("is_safe", "prediction").show(10)
println(f"模型准确率：$accuracy")

随机森林主要参数

参数	说明	示例值
`setNumTrees`	树的数量	100
`setMaxDepth`	每棵树的最大深度	10
`setMaxBins`	特征分割时的最大分箱数	32

Pipeline（管道）

Pipeline 把多个处理步骤串起来，形成一个工作流：

数据 → 特征组装 → 标准化 → 随机森林 → 预测结果

💡 Pipeline 的好处：
把整个流程封装成一个对象
训练和预测用同一个流程，不会出错
方便保存和加载

模型效果

准确率约 96%
效果良好，可直接保存用于后续使用

完整代码结构

scala

import org.apache.spark.sql.SparkSession
import org.apache.spark.ml.feature.{VectorAssembler, StandardScaler}
import org.apache.spark.ml.classification.RandomForestClassifier
import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.linalg.Matrix
import org.apache.spark.ml.stat.Correlation
import org.apache.spark.sql.Row

object water {
  def main(args: Array[String]): Unit = {
    // 1. 创建SparkSession
    val spark = SparkSession.builder()
      .appName("water")
      .master("local[*]")
      .enableHiveSupport()
      .getOrCreate()
    spark.sparkContext.setLogLevel("WARN")
    
    // 2. 读取数据
    val waterData = spark.read
      .option("header", "true")
      .option("inferSchema", "true")
      .csv("hdfs://master:8020/tipdm/data/waterQuality.csv")
    
    // 3. 探索性数据分析（略，见任务8.2）
    
    // 4. 特征组装（除标签列外）
    val newAssembler = new VectorAssembler()
      .setInputCols(waterData.columns.init)
      .setOutputCol("newFeatures")
    
    // 5. 数据标准化
    val scaler = new StandardScaler()
      .setInputCol("newFeatures")
      .setOutputCol("scaledFeatures")
      .setWithStd(true)
      .setWithMean(false)
    
    // 6. 随机森林模型
    val rf = new RandomForestClassifier()
      .setFeaturesCol("scaledFeatures")
      .setLabelCol("is_safe")
      .setPredictionCol("prediction")
      .setNumTrees(100)
      .setMaxDepth(10)
      .setMaxBins(32)
    
    // 7. 划分训练集和测试集
    val Array(trainingData, testData) = waterData.randomSplit(
      Array(0.8, 0.2), seed = 12345)
    
    // 8. 构建管道
    val pipeline = new Pipeline()
      .setStages(Array(newAssembler, scaler, rf))
    
    // 9. 训练
    val model = pipeline.fit(trainingData)
    
    // 10. 预测
    val predictions = model.transform(testData)
    
    // 11. 评估
    val evaluator = new MulticlassClassificationEvaluator()
      .setLabelCol("is_safe")
      .setPredictionCol("prediction")
      .setMetricName("accuracy")
    val accuracy = evaluator.evaluate(predictions)
    println(f"模型准确率：$accuracy")
  }
}

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

14.1 数据类问题

问题1：类型不匹配报错

现象：报 type mismatch 错误原因：特征列不是 Vector 类型，或者标签列类型不对解决：

用 VectorAssembler 把多列合并成 Vector
确保标签列是 Double 类型

问题2：LIBSVM 格式数据读不出来

现象：读取后数据不对原因：格式不对，索引从0开始了，或者分隔符不对解决：

LIBSVM 格式：标签索引1:值1 索引2:值2 ...
索引必须从 1 开始
用空格分隔

14.2 算法类问题

问题3：K-Means 结果每次不一样

现象：每次运行聚类结果不同原因：初始中心点是随机选的解决：设置随机种子 setSeed(1L)，保证结果可复现

问题4：ALS 预测结果有 NaN

现象：预测结果出现 NaN 原因：冷启动问题，测试集中有训练集没见过的用户/物品解决：设置冷启动策略

scala

model.setColdStartStrategy("drop")

问题5：模型准确率太低

现象：准确率很低 可能原因和解决：

数据质量差 → 先做数据清洗、特征工程
特征选择不对 → 分析相关性，选重要特征
参数不对 → 调参（树的数量、深度等）
数据不平衡 → 考虑过采样/欠采样

14.3 Pipeline 类问题

问题6：Pipeline 中各阶段的列名对不上

现象：报 column not found 错误原因：前一个阶段的输出列名和后一个阶段的输入列名不一致解决：检查每个 stage 的 setInputCol 和 setOutputCol，确保衔接正确

问题7：标签列参与了特征计算

现象：准确率异常高（100%）原因：把标签列也放进了特征列，模型直接"看到答案"了解决：确保特征列不包含标签列

scala

// 正确：除最后一列（标签列）外的所有列
.setInputCols(waterData.columns.init)

14.4 排错通用思路

先看数据：show() 一下，确认数据格式对不对
小数据测试：先用少量数据验证流程
逐步调试：每个 transform 后都 show 一下
检查列名：确保输入输出列名对应
看报错信息：仔细读错误提示，通常会告诉你哪列有问题

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

15.1 概念类（高频）

Q1：什么是机器学习？有哪几类？

机器学习是让计算机从数据中自动学习规律，并用规律来预测或决策的技术。按训练数据是否有标签分为三类：
监督学习：训练数据有标签，比如分类、回归
无监督学习：训练数据没有标签，比如聚类、关联规则
半监督学习：大部分没标签，小部分有标签

Q2：监督学习和无监督学习的区别？

监督学习：训练数据有标签（已知答案），目标是预测/分类，有明确对错
无监督学习：训练数据没有标签，目标是发现数据中的结构和模式，没有标准答案
例子：监督学习有线性回归、决策树；无监督学习有K-Means、FP-Growth

Q3：分类和聚类的区别？

分类是监督学习，有标签，已知类别，判断新数据属于哪类
聚类是无监督学习，没有标签，自动把相似的数据分到一组
分类的例子：判断是否合格、识别垃圾邮件
聚类的例子：用户分群、异常检测

Q4：Spark MLlib 中 spark.ml 和 spark.mllib 的区别？

spark.mllib：基于RDD的老API，Spark 2.0起进入维护模式，不再新增特性
spark.ml：基于DataFrame的新API，支持Pipeline工作流，官方推荐使用
新开发应该用 spark.ml 包

Q5：MLlib 中有哪些常见数据类型？

Vector（向量）：稠密和稀疏两种，最基础的特征表示
LabeledPoint（标注点）：标签 + 特征向量，用于监督学习
Matrix（矩阵）：二维数值数组
Rating：用户ID + 物品ID + 评分，用于推荐算法注意：RDD是Spark核心数据结构，不是MLlib特有的。

Q6：什么是 TF-IDF？

TF-IDF是一种文本特征提取方法，衡量词在文档中的重要程度。
TF（词频）：词在文档中出现的频率
IDF（逆文档频率）：词在整个语料库中的稀有程度
TF-IDF = TF × IDF 值越大，说明这个词对本文档越重要。在Spark中通过 HashingTF + IDF 实现。

Q7：Word2Vec 是什么？

Word2Vec是自然语言处理中的算法，把词转换成向量（词嵌入）。向量能表示词的语义信息，语义相近的词向量也相近。可以用来找同义词、做文本分类、推荐等。

Q8：标准化和归一化的区别？

都是特征处理方法，让不同特征在同一尺度上：
标准化（StandardScaler）：均值为0，标准差为1，适合近似正态分布的数据
归一化（MinMaxScaler）：缩放到[0,1]区间，适合需要固定范围的场景
Normalizer：按行归一化，每个样本的范数为1 前两个是按列（特征）处理，Normalizer是按行（样本）处理。

Q9：什么是过拟合？怎么解决？

过拟合就是模型在训练集上表现很好，但在测试集上表现很差。模型"记住"了训练数据的细节和噪声，而不是学到了通用规律。解决方法：
增加训练数据
正则化（regParam）
简化模型（比如决策树减小深度）
交叉验证

Q10：随机森林和决策树的关系？

随机森林由很多棵决策树组成。
每棵树用随机的样本和随机的特征训练
预测时所有树投票，少数服从多数
随机森林比单棵决策树更准确，不容易过拟合
但可解释性不如单棵决策树

15.2 算法类（中频）

Q11：K-Means 算法的原理？K值怎么选？

K-Means原理：
随机选K个中心点
每个点分配到最近的中心点
重新计算每个簇的中心点
重复2-3直到收敛
K值选择：
肘部法则：看K增加时误差下降的拐点
轮廓系数：选轮廓系数最高的K
业务经验：根据实际需求确定分几类

Q12：FP-Growth 是什么？支持度和置信度？

FP-Growth是关联规则挖掘算法，发现数据中频繁出现的模式。
支持度（Support）：某组物品一起出现的频率 = 出现次数/总订单数
置信度（Confidence）：买了A的人中同时买了B的比例 = P(AB)/P(A) 同时满足最小支持度和最小置信度的规则叫强关联规则。

Q13：协同过滤是什么？ALS是什么？

协同过滤是推荐算法，基于"相似用户喜欢相似物品"的思想。有基于用户和基于物品两种思路。
ALS（交替最小二乘法）是Spark实现的协同过滤算法：
把用户-物品评分矩阵分解成用户特征矩阵和物品特征矩阵
交替优化两个矩阵，直到收敛
支持显式评分和隐式反馈两种模式
有冷启动问题，可以用 setColdStartStrategy("drop") 处理

Q14：什么是 Pipeline？有什么好处？

Pipeline是Spark ML中的工作流，把多个处理阶段串起来。比如：特征组装 → 标准化 → 模型训练
好处：
把整个流程封装成一个对象，方便管理
训练和预测用同一个流程，保证一致性
方便保存和加载整个模型

15.3 实操类（高频）

Q15：怎么划分训练集和测试集？

用 randomSplit 方法：
scala
val Array(train, test) = data.randomSplit(Array(0.8, 0.2), seed = 12345)
Array(0.8, 0.2) 表示80%训练，20%测试
seed是随机种子，保证每次划分一样

Q16：分类问题怎么评估？常用什么指标？

用 MulticlassClassificationEvaluator 评估。常用指标：
准确率（accuracy）：预测正确的比例
精确率（precision）
召回率（recall）
F1值：精确率和召回率的调和平均

Q17：回归问题怎么评估？常用什么指标？

用 RegressionEvaluator 评估。常用指标：
MSE（均方误差）：误差平方的平均值
RMSE（均方根误差）：MSE开根号
MAE（平均绝对误差）：误差绝对值的平均值
R²（决定系数）：模型解释了多少方差，越大越好

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

scala
// 保存
model.save("modelPath")

// 加载
val model = LinearRegressionModel.load("modelPath")
PipelineModel 也可以保存和加载，整个工作流一起保存。

附录：习题解析

选择题解析

1、答案：A 解析：机器学习可以分为监督学习、无监督学习、半监督学习3种。

2、答案：C 解析：决策树算法主要用于分类或回归问题，K-Means算法用于聚类问题，朴素贝叶斯算法用于分类问题，只有FP-Growth算法属于关联规则挖掘。

3、答案：A 解析：K-Means算法为聚类算法，属于无监督学习算法，其他均为分类或回归算法，属于监督学习算法。

4、答案：B 解析：FCM和K-Means均属于聚类算法，FP属于关联规则算法，决策树属于分类/回归算法。

5、答案：C 解析：RDD是Spark的核心数据结构，不是MLlib特有的数据类型。

6、答案：C 解析：normL1方法用于计算向量的L1范数，而非求正态分布。

7、答案：D 解析：LabeledPoint用于监督学习，包含特征向量和标签，协同过滤推荐算法属于无监督学习，不需要用到LabeledPoint。

8、答案：D 解析：WithMean()只是StandardScaler方法的一个参数，并不是独立的标准化或归一化方法。

9、答案：D 解析：ALS是协同过滤推荐算法，输入的数据是RDD[Rating]，其中Rating包含三个字段：用户ID、物品ID、评分值。

10、答案：D 解析：HashingTF要求输入数据必须是可迭代的集合，如文本分词后的单词列表。

操作题解析

题目

手机价格预测，用随机森林回归模型预测手机价格区间。

实现代码

scala

import org.apache.spark.ml.evaluation.RegressionEvaluator
import org.apache.spark.ml.feature.{StandardScaler, VectorAssembler}
import org.apache.spark.ml.regression.RandomForestRegressor
import org.apache.spark.sql.SparkSession

object Phone {
  def main(args: Array[String]): Unit = {
    // 1. 初始化SparkSession
    val spark = SparkSession.builder()
      .appName("phone")
      .master("local[*]")
      .getOrCreate()
    spark.sparkContext.setLogLevel("WARN")
    
    // 2. 读取CSV数据，去空值，去重
    val phoneData = spark.read
      .option("header", "true")
      .option("inferSchema", "true")
      .csv("E:\\data\\phone.csv")
      .na.drop
      .distinct()
    
    // 3. 创建特征向量（除第一列id和最后一列price_range外）
    val vectorData = new VectorAssembler()
      .setInputCols(phoneData.columns.tail.init)
      .setOutputCol("features")
      .transform(phoneData)
    
    // 4. 数据标准化
    val scaler = new StandardScaler()
      .setInputCol("features")
      .setOutputCol("scaledFeatures")
      .setWithStd(true)
      .setWithMean(true)
    val scalerModel = scaler.fit(vectorData)
    val scaledData = scalerModel.transform(vectorData)
    
    // 5. 划分训练集和测试集（7:3）
    val Array(train, test) = scaledData.randomSplit(Array(0.7, 0.3))
    
    // 6. 创建随机森林回归模型
    val rf = new RandomForestRegressor()
      .setLabelCol("price_range")
      .setFeaturesCol("features")
    
    // 7. 训练 + 预测
    val model = rf.fit(train)
    val pre = model.transform(test)
    pre.select("id", "price_range", "prediction").show()
    
    // 8. 评估模型（MAE：平均绝对误差）
    val evaluator = new RegressionEvaluator()
      .setLabelCol("price_range")
      .setPredictionCol("prediction")
      .setMetricName("mae")
    val mae = evaluator.evaluate(pre)
    println(s"Mean Absolute Error (MAE) on test data = $mae")
  }
}

代码说明

代码	说明
`.na.drop`	删除空值
`.distinct()`	去重
`phoneData.columns.tail.init`	除第一列（id）和最后一列（price_range）外的所有列
`RandomForestRegressor`	随机森林回归（注意是Regressor不是Classifier）
`RegressionEvaluator`	回归评估器
`setMetricName("mae")`	用平均绝对误差评估

💡 注意：
分类用 RandomForestClassifier，评估用 MulticlassClassificationEvaluator
回归用 RandomForestRegressor，评估用 RegressionEvaluator
别搞混了！

笔记版本：V1.0 对应教材：《Spark大数据技术与应用（第3版）》人民邮电出版社 对应项目：项目8 饮用水源合格性预测——Spark MLlib机器学习算法库 最后更新：2026年8月

项目8：饮用水源合格性预测——Spark MLlib 机器学习算法库 ​

目录 ​

第一部分：项目背景与机器学习概述 ​

1.1 项目背景 ​

为什么需要机器学习预测？ ​

项目场景 ​

第二部分：机器学习基本概念 ​

2.1 什么是机器学习？ ​

机器学习的过程 ​

2.2 机器学习的分类 ​

按训练数据是否有标签分 ​

监督学习 vs 无监督学习 ​

2.3 常见机器学习算法类型 ​

第三部分：Spark MLlib 简介与发展 ​

3.1 什么是 MLlib？ ​

MLlib 包含的内容 ​

3.2 MLlib 的发展历史 ​

两个包的区别 ​

第四部分：MLlib 数据类型 ​

4.1 常见数据类型 ​

4.2 向量（Vector） ​

两种向量 ​

示例 ​

4.3 标注点（LabeledPoint） ​

第五部分：特征提取 ​

5.1 TF-IDF 算法 ​

什么是 TF-IDF？ ​

两个部分 ​

公式 ​

在 Spark MLlib 中的实现 ​

spark.mllib 包实现（基于RDD） ​

spark.ml 包实现（基于DataFrame，推荐） ​

5.2 Word2Vec 算法 ​

什么是 Word2Vec？ ​

spark.mllib 包实现 ​

spark.ml 包实现 ​

5.3 统计方法 ​

spark.mllib 包（基于RDD） ​

spark.ml 包（基于DataFrame） ​

第六部分：数据特征处理（标准化/归一化） ​

6.1 为什么要做特征处理？ ​

6.2 三种常用方法对比 ​

准备数据 ​

6.3 Normalizer（行归一化） ​

作用 ​

示例 ​

参数 ​

6.4 StandardScaler（标准化） ​

作用 ​

公式 ​

示例 ​

参数 ​

6.5 MinMaxScaler（最小-最大归一化） ​

作用 ​

公式 ​

示例 ​

参数 ​

6.6 三种方法总结 ​

第七部分：回归算法 ​

7.1 线性回归 ​

什么是线性回归？ ​

公式 ​

数据格式：LIBSVM 格式 ​

示例代码 ​

主要参数 ​

7.2 逻辑回归 ​

什么是逻辑回归？ ​

与线性回归的区别 ​

包位置 ​

第八部分：分类算法 ​

8.1 朴素贝叶斯 ​

什么是朴素贝叶斯？ ​

原理 ​

示例：根据天气判断是否打球 ​

8.2 决策树 ​

什么是决策树？ ​

特点 ​

包位置 ​

原理 ​

第九部分：聚类算法（K-Means） ​

项目8：饮用水源合格性预测——Spark MLlib 机器学习算法库

目录

第一部分：项目背景与机器学习概述

1.1 项目背景

为什么需要机器学习预测？

项目场景

第二部分：机器学习基本概念

2.1 什么是机器学习？

机器学习的过程

2.2 机器学习的分类

按训练数据是否有标签分

监督学习 vs 无监督学习

2.3 常见机器学习算法类型

第三部分：Spark MLlib 简介与发展

3.1 什么是 MLlib？

MLlib 包含的内容

3.2 MLlib 的发展历史

两个包的区别

第四部分：MLlib 数据类型

4.1 常见数据类型

4.2 向量（Vector）

两种向量

示例

4.3 标注点（LabeledPoint）

第五部分：特征提取

5.1 TF-IDF 算法

什么是 TF-IDF？

两个部分

公式

在 Spark MLlib 中的实现

spark.mllib 包实现（基于RDD）

spark.ml 包实现（基于DataFrame，推荐）

5.2 Word2Vec 算法

什么是 Word2Vec？

spark.mllib 包实现

spark.ml 包实现

5.3 统计方法

spark.mllib 包（基于RDD）

spark.ml 包（基于DataFrame）

第六部分：数据特征处理（标准化/归一化）

6.1 为什么要做特征处理？

6.2 三种常用方法对比

准备数据

6.3 Normalizer（行归一化）

作用

示例

参数

6.4 StandardScaler（标准化）

作用

公式

示例

参数

6.5 MinMaxScaler（最小-最大归一化）

作用

公式

示例

参数

6.6 三种方法总结

第七部分：回归算法

7.1 线性回归

什么是线性回归？

公式

数据格式：LIBSVM 格式

示例代码

主要参数

7.2 逻辑回归

什么是逻辑回归？

与线性回归的区别

包位置

第八部分：分类算法

8.1 朴素贝叶斯

什么是朴素贝叶斯？

原理

示例：根据天气判断是否打球

8.2 决策树

什么是决策树？

特点

包位置

原理

第九部分：聚类算法（K-Means）