Spark 机器学习——凸优化算法、线性回归算法、FPGrowth关联规则算法、决策树等

本帖最后由 a87758133 于 2019-5-16 14:09 编辑
问题导读：

1、凸优化是什么？
2、如何实现从经验风险最小化到结构经验最小化？
3、线性回归算法的数学模型是怎样的？
4、FPGrowth挖掘过程是怎样的？
5、协同过滤的步骤有哪些？

第4章凸优化算法
不严格的说，凸优化就是在标准优化问题的范畴内，要求目标函数和约束函数是凸函数的一类优化问题。

注意：中国大陆数学界某些机构关于函数凹凸性定义和国外的定义是相反的。Convex Function 在某些中国大陆的数学书中指凹函数。Concave Function 指凸函数。但在中国大陆涉及经济学的很多书中，凹凸性的提法和其他国家的提法是一致的，也就是和数学教材是反的。举个例子，同济大学高等数学教材对函数的凹凸性定义与本条目相反，本条目的凹凸性是指其上方图是凹集或凸集，而同济大学高等数学教材则是指其下方图是凹集或凸集，两者定义正好相反。

4.1 梯度下降

4.2 牛顿法

4.3 拟牛顿法

4.4 BFGS 算法

第5章 L1、L2 正则化

5.1 从经验风险最小化到结构经验最小化

5.2 范数与正则项

5.3 贝叶斯先验

第6章线性回归算法
6.1 数学模型
  线性回归是利用被称为线性回归方程的最小平方函数对一个或者多个自变量和因变量之间关系进行建模的一种回归分析。这种函数式一个或者多个被称为回归系数的模型参数的线性组合。
  在统计学中，线性回归(Linear Regression) 是利用称为线性回归方程的最小平方函数对一个或多个自变量和因变量之间关系进行建模的一种回归分析。这种函数是一个或多个称为回归系数的模型参数的线性组合。
  回归分析中，只包括一个自变量和一个因变量，且二者的关系可用一条直线近似表示，这种回归分析称为一元线性回归分析。如果回归分析中包括两个或两个以上的自变量，且因变量和自变量之间是线性关系，则称为多元线性回归分析。

6.2 线性回归的应用
  通过大量样本的试验学习到线性函数，然后根据新的样本的特征数据，预测结果。

6.2 Spark MLlib 实现
示例代码：

[mw_shl_code=scala,true]package com.atguigu.mllib

import org.apache.log4j.{Level, Logger}
import org.apache.spark.mllib.regression.{LinearRegressionModel, LinearRegressionWithSGD}
import org.apache.spark.mllib.util.MLUtils
import org.apache.spark.{SparkConf, SparkContext}

object LinearRegression {

  def main(args: Array[String]): Unit = {

// 屏蔽日志
Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

// 创建 SparkContext
val conf = new SparkConf().setMaster("local[4]").setAppName("LinearRegression")
val sc = new SparkContext(conf)

// 加载数据样本
val path = "D:\\learn\\JetBrains\\workspace_idea\\spark\\doc\\data.txt";

// 通过提供的工具类加载样本文件
val data = MLUtils.loadLibSVMFile(sc, path).cache()

// 或者通过 RDD 转换加载
/*
val data = sc.textFile(path).map { line =>
   val parts = line.split(' ')
   LabeledPoint(parts(0).toDouble, Vectors.dense(parts.tail.map(_.split(":")(1).toDouble)))
}.cache()
*/

// 迭代次数
val numIterations = 100
// 梯度下降步长
val stepSize = 0.00000001
// 训练模型
val model = LinearRegressionWithSGD.train(data, numIterations, stepSize)

// 模型评估
val valuesAndPreds = data.map { point =>
   // 根据模型预测 Label 值
   val prediction = model.predict(point.features)
   println(s"【真实值】：${point.label}    ;【预测值】：${prediction}")
   (point.label, prediction)
}

// 求均方误差
val MSE = valuesAndPreds.map { case (v, p) => math.pow((v - p), 2) }.mean()
println("训练模型的均方误差为 = " + MSE)

// 保存模型
model.save(sc, "target/tmp/scalaLinearRegressionWithSGDModel")
// 重新加载模型
val sameModel = LinearRegressionModel.load(sc, "target/tmp/scalaLinearRegressionWithSGDModel")

sc.stop()
  }
}[/mw_shl_code]
输出结果如下：
[mw_shl_code=text,true]【真实值】：-9.490009878824548    ;【预测值】：7.708618803157486E-9
【真实值】：-1.1838791995691869    ;【预测值】：1.1953486679306154E-8
【真实值】：0.2577820163584905    ;【预测值】：-2.460347544748461E-9
【真实值】：-1.5856680515554806    ;【预测值】：1.43032697303707E-8
【真实值】：-13.039928064104615    ;【预测值】：-2.909264331661566E-10
【真实值】：-4.438869807456516    ;【预测值】：-4.3817348405900043E-10
【真实值】：-17.428674570939506    ;【预测值】：7.864237801670372E-9
【真实值】：-19.782762789614537    ;【预测值】：2.749919462104237E-9
......
......
训练模型的均方误差为 = 106.31223022762704[/mw_shl_code]
第7章 FPGrowth 关联规则算法
7.1算法思想
  FPGrowth 算法通过构造一个 FPTree 树结构来压缩数据记录，使得挖掘频繁项集只需要扫描两次数据记录，而且该算法不需要生成候选集合，所以效率会比较高。如何从购物篮里面发现尿布+啤酒的最佳组合。 >   我们以以下数据集为例：

  注意：牛奶、面包叫做项，{ 牛奶、面包 } 叫做项集。项集出现的次数叫做支持度。T* 表示用户每次的购物清单。
  FPGrowth挖掘过程如下图所示：

7.2 Spark MLlib 实现
示例代码如下：

[mw_shl_code=scala,true]package com.atguigu.mllib

import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.mllib.fpm.FPGrowth

object FPGrowth extends App {

  // 屏蔽日志
  Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
  Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

  // 创建 SparkContext
  val conf = new SparkConf().setMaster("local[4]").setAppName("FPGrowth")
  val sc = new SparkContext(conf)

  // 加载数据样本
  val path = "D:\\learn\\JetBrains\\workspace_idea\\spark\\doc\\fpgrowth.txt";
  // 创建交易样本
  val transactions = sc.textFile(path).map(_.split(" ")).cache()

  println(s"交易样本的数量为： ${transactions.count()}")

  // 最小支持度 [0,1]
  val minSupport = 0.4

  // 计算的并行度
  val numPartition = 2

  // 训练模型
  val model = new FPGrowth()
.setMinSupport(minSupport)
.setNumPartitions(numPartition)
.run(transactions)

  // 打印模型结果
  println(s"经常一起购买的物品集的数量为： ${model.freqItemsets.count()}")
  model.freqItemsets.collect().foreach { itemset =>
println(itemset.items.mkString("[", ",", "]") + ", " + itemset.freq)
  }

  sc.stop()
}
[/mw_shl_code]
输出结果如下：
[mw_shl_code=text,true]交易样本的数量为： 6
经常一起购买的物品集的数量为： 18
[t], 3
[t,x], 3
[t,x,z], 3
[t,z], 3
[ s], 3
[s,x], 3
[z], 5
[y], 3
[y,t], 3
[y,t,x], 3
[y,t,x,z], 3
[y,t,z], 3
[y,x], 3
[y,x,z], 3
[y,z], 3
[x], 4
[x,z], 3
[r], 3
[/mw_shl_code]
第8章协同过滤推荐算法
8.1 算法思想
  比如你想看一个电影，但是不知道具体看哪一部，你会怎么做？有两种办法，一种是问问周围兴趣相似的朋友，看看他们最近有什么好的电影推荐。另外一种是看看电影的相似程度，比如都喜欢看僵尸片，那就会找电影名带有僵尸、丧尸之类的电影。
  协同过滤算法就是基于上面的思想，主要包含基于用户的协同过滤推荐算法以及基于物品的协同过滤推荐算法。
  实现协同过滤，一般需要几个步骤：
  1、收集用户偏好。
  2、找到相似的用户或者物品。
  3、计算推荐。

  协同过滤算法主要用于推荐系统，推荐系统是信息过载所采用的措施，面对海量的数据信息，从中快速推荐出符合用户特点的物品。一些人的“选择恐惧症”、没有明确需求的人。
  • 解决如何从大量信息中找到自己感兴趣的信息。
  • 解决如何让自己生产的信息脱颖而出，受到大众的喜爱。

8.2 相似性度量

8.3 Spark MLlib 实现
推荐数据的准备

协同过滤推荐架构

示例代码如下：
[mw_shl_code=scala,true]/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements.  See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License.  You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

// scalastyle:off println
package com.spark.mllib

import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}

import org.apache.spark.mllib.recommendation.ALS
import org.apache.spark.mllib.recommendation.MatrixFactorizationModel
import org.apache.spark.mllib.recommendation.Rating

object Recommendation {
  def main(args: Array[String]): Unit = {

// 屏蔽日志
Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

// 创建 SparkContext
val conf = new SparkConf().setMaster("local[4]").setAppName("CollaborativeFiltering")
val sc = new SparkContext(conf)

// 加载数据
val path = "D:\\learn\\JetBrains\\workspace_idea\\spark\\doc\\test.data"
val data = sc.textFile(path)
val ratings = data.map(_.split(',') match { case Array(user, item, rate) =>
   Rating(user.toInt, item.toInt, rate.toDouble)
})

// 训练模型
val rank = 50
val numIterations = 10
val model = ALS.train(ratings, rank, numIterations, 0.01)

// 准备用户数据
val usersProducts = ratings.map { case Rating(user, product, rate) =>
   (user, product)
}

// 生成推荐结果
val predictions =
   model.predict(usersProducts).map { case Rating(user, product, rate) =>
      ((user, product), rate)
   }

// 对比结果
val ratesAndPreds = ratings.map { case Rating(user, product, rate) =>
   ((user, product), rate)
}.join(predictions)

// 生成均方误差
val MSE = ratesAndPreds.map { case ((user, product), (r1, r2)) =>
   println(s"【用户】：${user}  【物品】：${product}  【真实值】：${r1}  【预测值】：${r2}")
   val err = (r1 - r2)
   err * err
}.mean()

println("预测的均方误差为 = " + MSE)

// 保存模型
model.save(sc, "target/tmp/myCollaborativeFilter")
// 加载模型
val sameModel = MatrixFactorizationModel.load(sc, "target/tmp/myCollaborativeFilter")

sc.stop()
  }
}
[/mw_shl_code]
输出结果如下：
[mw_shl_code=text,true]【用户】：4  【物品】：4  【真实值】：5.0  【预测值】：4.996434384228974
【用户】：2  【物品】：1  【真实值】：5.0  【预测值】：4.9967879248395715
【用户】：1  【物品】：1  【真实值】：5.0  【预测值】：4.9967879248395715
【用户】：4  【物品】：2  【真实值】：5.0  【预测值】：4.996434384228974
【用户】：1  【物品】：4  【真实值】：1.0  【预测值】：1.0001174801070112
【用户】：4  【物品】：1  【真实值】：1.0  【预测值】：1.0001881676270354
【用户】：2  【物品】：2  【真实值】：1.0  【预测值】：1.0001174801070112
【用户】：3  【物品】：1  【真实值】：1.0  【预测值】：1.0001881676270354
【用户】：2  【物品】：4  【真实值】：1.0  【预测值】：1.0001174801070112
【用户】：3  【物品】：4  【真实值】：5.0  【预测值】：4.996434384228974
【用户】：2  【物品】：3  【真实值】：5.0  【预测值】：4.9967879248395715
【用户】：3  【物品】：2  【真实值】：5.0  【预测值】：4.996434384228974
【用户】：1  【物品】：2  【真实值】：1.0  【预测值】：1.0001174801070112
【用户】：3  【物品】：3  【真实值】：1.0  【预测值】：1.0001881676270354
【用户】：4  【物品】：3  【真实值】：1.0  【预测值】：1.0001881676270354
【用户】：1  【物品】：3  【真实值】：5.0  【预测值】：4.9967879248395715
预测的均方误差为 = 5.7700628235600924E-6
[/mw_shl_code]
第9章决策树
9.1 算法思想
  决策树（Decision Tree）是一种基本的分类与回归方法。决策树模型呈树形结构，在分类问题中，表示基于特征对实例进行分类的过程。它可以认为是 if-then 规则的集合，也可以认为是定义在特征空间与类空间上的条件概率分布。相比朴素贝叶斯分类，决策树的优势在于构造过程不需要任何领域知识或参数设置，因此在实际应用中，对于探测式的知识发现，决策树更加适用。
  分类决策树模型是一种描述对实例进行分类的树形结构。决策树由结点和有向边组成。结点有两种类型：内部节点和叶节点，内部节点表示一个特征或属性，叶节点表示一个类。
  分类的时候，从根节点开始，对实例的某一个特征进行测试，根据测试结果，将实例分配到其子结点；此时，每一个子结点对应着该特征的一个取值。如此递归向下移动，直至达到叶结点，最后将实例分配到叶结点的类中。
  举一个通俗的例子，各位立志于脱单的单身男女在找对象的时候就已经完完全全使用了决策树的思想。假设一位母亲在给女儿介绍对象时，有这么一段对话：
  母亲：给你介绍个对象。
  女儿：年纪多大了？
  母亲：26。
  女儿：长的帅不帅？
  母亲：挺帅的。
  女儿：收入高不？
  母亲：不算很高，中等情况。
  女儿：是公务员不？
  母亲：是，在税务局上班呢。
  女儿：那好，我去见见。
  这个女生的决策过程就是典型的分类决策树。相当于对年龄、外貌、收入和是否公务员等特征将男人分为两个类别：见或者不见。假设这个女生的决策逻辑如下：

上图完整表达了这个女孩决定是否见一个约会对象的策略，其中绿色结点（内部结点）表示判断条件，橙色结点（叶结点）表示决策结果，箭头表示在一个判断条件在不同情况下的决策路径，图中红色箭头表示了上面例子中女孩的决策过程。这幅图基本可以算是一棵决策树，说它 “基本可以算” 是因为图中的判定条件没有量化，如收入高中低等等，还不能算是严格意义上的决策树，如果将所有条件量化，则就变成真正的决策树了。

9.2 决策树模型的两种解释
  分类决策树模型是一种描述对实例进行分类的树形结构。决策树由结点和有向边组成。结点有两种类型：内部结点和叶节点。内部结点表示一个特征或属性，叶节点表示一个类。

9.2.1 决策树与 if-then 规则
  可以将决策树看成一个 if-then 规则的集合。即由决策树的根结点到叶节点的每一条路径构建一条规则；路径上内部结点的特征对应着规则的条件，而叶结点的类对应着规则的结论。
  决策树的路径或其对应的 if-then 规则集合的重要性质：互斥且完备（每一个实例都被一条路径或一条规则所覆盖，且只被一条路径或一条规则所覆盖，这里的覆盖是指实例的特征与路径上的特征一致或实例满足规则的条件）

9.2.2 决策树与条件概率分布
  决策树还表示给定特征条件下类的条件概率分布，它定义在特征空间的一个划分。将特征空间划分为互不相交的单元，并在每个单元定义一个类的概率分布就构成了一个条件概率分布。决策树的每一条路径对应于划分中的一个单元。
  假设 X 为表示特征的随机变量，Y 为表示类的随机变量，那么这个条件概率分布可以表示为 P(X|Y)，各叶结点上的条件概率往往偏向于某一个类，即属于某一类的概率越大。决策树分类时将该结点的实例强行分到条件概率大的那一类去。

9.3 特征选择

9.4 Spark MLlib 实现
示例代码如下：

[mw_shl_code=scala,true]package com.atguigu.mllib

import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.mllib.tree.DecisionTree
import org.apache.spark.mllib.tree.model.DecisionTreeModel
import org.apache.spark.mllib.util.MLUtils

object DecisionTreeApp extends App {
  // 屏蔽日志
  Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
  Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

  // 创建 SparkContext
  val conf = new SparkConf().setMaster("local[4]").setAppName("DecisionTree")
  val sc = new SparkContext(conf)

  val path = "D:\\learn\\JetBrains\\workspace_idea\\spark\\sparkmllib_decision_tree\\src\\main\\resources\\data.txt"
  // 加载数据文件
  val data = MLUtils.loadLibSVMFile(sc, path)
  // 将数据集切分为 70% 的训练数据集和 30% 的测试数据集
  val splits = data.randomSplit(Array(0.7, 0.3))
  val (trainingData, testData) = (splits(0), splits(1))

  // 训练决策树模型
  // Empty categoricalFeaturesInfo indicates all features are continuous.
  val numClasses = 2
  val categoricalFeaturesInfo = Map[Int, Int]()
  val impurity = "gini"
  val maxDepth = 5
  val maxBins = 32

  val model = DecisionTree.trainClassifier(trainingData, numClasses, categoricalFeaturesInfo,
impurity, maxDepth, maxBins)

  // 评估模型
  val labelAndPreds = testData.map { point =>
val prediction = model.predict(point.features)
(point.label, prediction)
  }

  val testErr = labelAndPreds.filter(r => r._1 != r._2).count().toDouble / testData.count()
  println("分类错误度 = " + testErr)
  println("训练的决策树模型:\n" + model.toDebugString)

  // 保存决策树模型
  model.save(sc, "target/tmp/myDecisionTreeClassificationModel")
  // 重新读取决策树模型
  val sameModel = DecisionTreeModel.load(sc, "target/tmp/myDecisionTreeClassificationModel")

  sc.stop()
}
[/mw_shl_code]
输出结果如下：
[mw_shl_code=text,true]分类错误度 = 0.06896551724137931
训练的决策树模型:
DecisionTreeModel classifier of depth 2 with 5 nodes
  If (feature 406 <= 20.0)
If (feature 99 <= 0.0)
Predict: 0.0
Else (feature 99 > 0.0)
Predict: 1.0
  Else (feature 406 > 20.0)
Predict: 1.0
[/mw_shl_code]
第10章随机森林算法