spark mllib和spark ml机器学习基础知识_sparkmil

spark机器学习

Spark MLib完整基础入门教程 - y-z-f - 博客园 (cnblogs.com)

参考

spark 机器学习简介

机械学习是一门人工智能的科学，用于研究人工智能，强调算法，经验，性能

开发者任务：spark基础+了解机器学习原理+相关参数含义

millib：

分类 回归 聚类 协同过滤 降维

特征化：特则提取 转化 降维 选择公交

管道：构建评估调整机器学习管道

持久性：保存和加载 算法，模型和管道

实用工具：线代(Breeze,jblas库) 统计 数据处理 的工具

spark:millib：基于RDD原始算法api（从spark2.0维护模式 3.0移除）

spark.ml:基于DataFrame的高层次API

ML Pipeline API：数据处理 特征转换 正则化 多个机器学习算法联合==》单一完整的机器学习流水线

spark.mllib

RDD常见函数

来源

数据类型

本地向量（单机模式存储本地向量和本地矩阵，一个或多个RDD分布式矩阵）

整形+索引值+浮点型元素

org.apache.spark.mllib.linalg.Vector

稠密向量DenseVector：双精度浮点型数组[1.0,0.0,3.0]

稀疏向量SparseVector：n个+整形索引数组+双精度浮点型元素(3,[0,1],[1.0,3.0])

//1. 导包org.apache.spark.mllib.linalg.{Vector,Vectors} 
import org.apache.spark.millib.linalg.Vectors
import org.apache.spark.sql.SparkSession
//2.创建变量Vectors.dense/sparse 参数为数组Array() 元组Seq()
object name{
    def main(args:Array[String]):Unit={
        val spark=SparkSession.builder().master("local[*]").appname("name").getOrCreate()
        val d1=Vectors.dense([1.3,2.0,3.0])
        val s1=Vectors.sparse(5,Array(1,4),(3.0,3.0))
        print(d1)
        print(s1)
    }
    
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

标注点（表示监督学习中的一个训练样本）:带标签的本地向量

用于回归Regression和分类Classification（正1负0 BNM,./.,MBNVBCXQ1YTUJHGFNBVC X）问题上

双精度浮点型标签+本地向量

//1.导包 org.apache.spark.mllib.regression.LabeledPoint  org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.linalg.Vectors
//2.创建变量LabeledPoint(1.0,Vectors.dense/sparse())
LabeledPoint(1.0,Vectors.dense(...))
1
2
3
4
5

mllib提供读取LIBSVM 格式数据 LIBSVM LIBLINEAR等机器学习库

//1.导包org.apache.spark.mllib.unil.MLUtils
import org.apache.spark.mllib.util.MLUtils
import org.apache.spark.millin.linalg.Vectors
//2.读loadLibSVMFile(sc,"*.txt") label index:value index:value index:value
val sc=SparkContext(new SparkConf().setMaster("").setAppName(""))
val svm1=MLUtils.loadLibSVMFile(sc,"*.txt")
svm1.collect().head()
svm1.collect().foreach(println)
1
2
3
4
5
6
7
8

本地矩阵(单机存储)

稠密矩阵DenseMatrix:列优先的双精度数组

稀疏矩阵Sparse Matrix:列优先的csc模式存储

//1.导包 org.apache.spark.mllib.linalg.Matricsx

//2.创建 Matrix.dense(行数，列数，元素)
Matrices.dense(3,2,Array(1.0,3.0,5.0,2.0,4.0,6.0))
/// Matrices.sparse(行数，列数，列索引，行索引，元素)
Matrices.sparse(3,2,Array(0,1,3),Array(0,2,1),Array(9,6,8))
1
2
3
4
5
6

分布式矩阵

长整型行索引+双精度浮点型元素

行矩阵RowMatrix

每一行输一个本地向量

numRows/numCols():获取行数/列数

rows:获取行 new RowMatrix(…).rows.foreach(println)

computeColumnSummaryStatistics():统计评价信息：QR分解 SVD分解 PCA分解

​ (得到每一列最大值)max/count/variance/mean/normL1

//1. 搭配包 org.apache.spark.linalg.distributed.RowMatrix
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.linalg.distributed.RowMatrix
//2.创建new RowMatriix(rows)
val dv1=Vectors.dense(1.0,2.0,3.0)
val dv2=Vectors.dense(2.0,3.0,4.0)
val mat=new RowMatrix(sc.parallelize(Array(dv1,dv2)))
mat.numRows()
mat.numCols()
mat.rows.foreach(println)
//3.评价
val summary=mat.computeColumnSummaryStatistics()
summary.count
summary.max/variance/mean/normL1
1
2
3
4
5
6
7
8
9
10
11
12
13
14

索引行矩阵IndexedRowMatrix

每一行一个有意义的行索引值存在IndexedRow组成(sc.parallelize())的RDD里

IndexedRow(索引，本地向量)

//1.导包 IndexedRow IndexedRowMatrix
import org.apache.spark.mllib.linalg.distrubuted.{IndexedRow,IndexedRowMatrix}
import org.apache.spark.{SparkConf,SparkContext}
import org.apache.spark.mllib.linalg.Vectors
//2.新建
val sc=new SparkContext(new SparkConf().setMaster("local[*]").setAppName("test"))
val id1=IndexedRow(1,Vectors.dense(1.0,2.0,3.0))
val id2=IndexedRow(2,Vectors.dense(2.0,3.0,4.0))
val idm=new IndexedRowMatrix(sc.parallelize(Array(id1,id2)))
idm.rows.foreach(println)
//3.获取
1
2
3
4
5
6
7
8
9
10
11

坐标矩阵CoordinateMatrix

基于矩阵项(MatrixEntry)构成的 (i行索引，j列索引，value元素)

new MatrixENtry(indexX,indexY,value)

transpose() 转置

toIndexedRowMatrix()转换成索引行矩阵IndexedRowMatrix

用于矩阵维度很大但稀疏的时候

//
import org.apache.spark.mllib.linalg.distrubuted.{CoordinateMatrix,MatrixEntry}
val en2=new MatrixEntry(0,1,0.5)
val en1=new MatrixEntry(2,2,1.8)
val ens=new CoordinateMatrix(sc.parallelize(Array(en1,en2)))
ens.entries.foreach(println)
val transens=ens.transpose().entries.foreach(println)
transens.toIndexedRowMatrix().rows.foreach(println)
1
2
3
4
5
6
7
8

分块矩阵 Block Matrix

基于矩阵块MatrixBlock((索引int,索引int),子矩阵Matrix)

validate:确认是否创建成功

索引行IndexedRowMatrix/坐标矩阵CoordinateMatrix.toBlockMatrix(尺寸) :分块转为分块矩阵

rowsPerBlock colPerBlock:大小/尺寸

import org.apache.spark.mllib.linalg.distributed.{BlockMatrix,CoordinateMatrix, MatrixEntry}

val ent1 = new MatrixEntry(0,0,1)
val ent2 = new MatrixEntry(1,1,1)
val ent3 = new MatrixEntry(2,0,-1)
val ent4 = new MatrixEntry(2,1,2)
val ent5 = new MatrixEntry(2,2,1)
val ent6 = new MatrixEntry(3,0,1)
val ent7 = new MatrixEntry(3,1,1)
val ent8 = new MatrixEntry(3,3,1)
val coordMat=new CoordinateMatrix(sc.parallelize(Array(ent1,ent2,ent3,ent4,ent5,ent6,ent7,ent8)))
val mat=coordMat.toBlockMatrix(2,2).cache()
mat.validate()
mat.toLocalMatrix
mat.numColBlocks
mat.numRowBLocks
val ata=mat.transpose.multiply(mat)  //转置相乘
ata.toLocalMatrix
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

数据统计工具

数据分析师先查看数据集的基本情况：汇总统计或概要性统计

一般的概要性统计用于概括一系列观测值，包括位置或集中趋势（比如算术平均值、中位数、众数和四分位均值），展型（比如四分位间距、绝对偏差和绝对距离偏差、各阶矩等），统计离差，分布的形状，依赖性等

spark.mllib库也提供了一些其他的基本的统计分析工具，包括相关性、分层抽样、假设检验，随机数生成等

• 概括统计 summary statistics

  org.apache.spark.mllib.stat.Statistics

  Statistics.colStats(Rdddata).count/mean/variance/max/min/normL1/normL2/numNonzeros非零向量个数

  import org.apache.spark.millib.linalg.Vectors
  import org.apache.spark.millib.stat.Statistics
  val sdata=sc.textFile("iris.data").map(_.split(',')).map(
      p=>Vectors.dense(p(0).toDouble,p(1).toDouble,p(2).toDouble,p(3).toDouble))
  val summary=Statistics.colStats(sdata)
  summary.count/max/min/normL1/normL2/numNonzeros/variance/mean
  1
  2
  3
  4
  5
  6

• 相关性 correlations

  Statistics.corr(x,y,“pearson”) ： 输出一个RDD[Vector]

  ​ 相关矩阵第i行第j列的元素是原矩阵第i列和第j列的相关系数

  Statistics.corr(data,“pearson”) 输出一个RDD[double]

  pearson皮尔逊系数[-1,1] :[-1,0)负相关 (0,1]正相关

  

  Spearman斯皮尔曼等级相关系数：用于排序

  

  import org.apache.spark.mllib.stat.Statistics
  import org.apache.spark.mllib.linalg._
  import org.apache.spark.{SparkContext,SparkConf}
  data=sc.textFile("iris.data").map(_.split(","))
  x=data.map(p=>{
      p(0).toDouble
  })
  y=data.map(p=>{
      p(1).toDouble
  })
  sdata=data.map(p=>{
      Vectors.dense(p(0).toDouble,p(1).toDouble)
  })
  println(Statistics.corr(x,y,"pearson"))
  println(Statistics.corr(data,"perason")) 
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15

• 分层抽样 Stratified sampling

  sampleBykeyExact和sampleByKey 区别：

  ​ sampleBykey每次一定概率掷硬币

  ​ sampleBykeyExact:全量数据作采样计算 每个类别都有类count*类比例个

  rdddata.sampleByKey(withReplacement=false,fractions,1)

  fractions:Map(“name1”->0,“name2”->0.4) 不同key的概率

  seed:随机种子：1

  withReplacement:是否放回

  import org.apache.spark.mllib.linalg.Vectors
  import org.apache.spark.{SparkConf,SparkContext}
  val data=sc.makeRDD(Array(
            ("female","Lily"),
            ("female","Lucy"),
            ("female","Emily"),
            ("female","Kate"),
            ("female","Alice"),
            ("male","Tom"),
            ("male","Roy"),
            ("male","David"),
            ("male","Frank"),
            ("male","Jack")))
  val fractions=Map("female"->0.6,"male"->0.4) 
  //female占抽样总数的0.6 male占0.4所以demale是5*0.6 和5*0.4
  data.sampleByKey(data,fractions,seed=1).collect().foreach(println)
  data.sampleByKeyExact(data,fractions,seed=1).collect().foreach(println)
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17

• 假设检验 hypothesis testing

  皮尔森卡方检测(Pearson’s chisquaerd tests):适配度检定和独立性检定

  Statistics.chiSqTest(value1[,value2])：一个值是适合度检验，两个是独立性检验

  Statistics.kolmogorovSmirnovTest(test,“norm”,0,1)

import org.apache.spark.SparkContext
import org.apache.spark.mllib.linalg._
import org.apche.spark.regression.LabeledPoint
import org.apache.spark.mllib.stat.Statistics._

val data=sc.textFile("iris.data")//原数据
val v1=data
	.map(_.split(",")) 
	.map(p=>Vectors.dense(p(0).toDouble,p(1).toDouble,p(2).toDouble,p(3).toDouble)).first //第一行数据
val v2=data
	.map(_.split(",")) .
	map(p=>Vectors.dense(p(0).toDouble,p(1).toDouble,p(2).toDouble,p(3).toDouble)).take(2).last //第二行数据
Statistics.chiSqTest(v1)  //得到均匀分配的假设，检测指标数据是否均匀分配 p=0.133
Statistics.chiSqTest(Matrices.dense(2,2,Array(v1(0),v1(1),v2(0),v2(1))))
Statistics.chiSqTest(v1,v2)

 val obs = data.map{ line =>
 	val parts = line.split(',')
      LabeledPoint(if(parts(4)=="Iris-setosa") 0.toDouble else if (parts(4)=="Iris-versicolor") 1.toDouble else
           2.toDouble, Vectors.dense(
               parts(0).toDouble,parts(1).toDouble,
               parts(2).toDouble,parts(3).toDouble))}
Statistics.chiSqTest(obs).foreach(println)

val KolmogorovTest=data.map(_.spli(",")).map(p=>p(0).toDouble)
println(Statistics.KolmogorovSmirnovTest(KolmogorovTest,"norm",0,1))
val mydf=>Double=(p=>p*2)
println(Statistics.kolmogorovSmirnovTest(test,mydf))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

• 随机数生成 random data generation

  RandomRDDs

  正态分布 泊松分布 均匀分布下的double/vector RDDS

  import org.apache.spark.mllib.random.RandomRDDS._
  val u=normalRDD(sc,10000000L,10) 
  //生成1000000个服从正太分配N(0,1)的RDD[Double]  并却在10个分区中
  u.map(x=>1.0+2.0*x)  //转成正态分布
  
  1
  2
  3
  4
  5

• 核密度估计 Kernel density estimation

  根据一致的样本估计未知的密度 属于非参数检验方法之一

  先观察某一事物分布–》某个数在观察中出现了，就认为这个数密度大，和这个数渐进的数的密度也大，离他远的数密度小

  建立模型：val model=new kernelDensity().setSample(data).setBandwidth(3.0)

  评估密度:model.estimate(Array(data))

//1.导包 org.apache.spark.mllib.stat.{MultivariateStatisticalSummary,Statistics}
import org.apache.spark.mllib.stat.KernelDensity
import org.apache.spark.rdd.RDD
val test=sc.textFile("").map(_.split(",")).map(p=>p(0).toDouble)
val kd=new KernelDensity().setSample(test).setBandwidth(3.0) //setBandwidth高斯核宽度 高斯核标准差
kd.estimate(Array((-1.0,2.0,5.0,5,9)))  //评估概率
1
2
3
4
5
6

奇异值分解-降维一

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Fv26IgFe-1663928932137)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220921215442833.png)]

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-9S1b9oSH-1663928932138)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220921215529105.png)]

rm.computeSVD(3[,computeU=True]) 保留3个奇异值

computeU=True:保留U成员

import org.apache.spark.mllib.linalg.distributed.RowMatrix
val value=sc.textFile("a.mat").map(_.split(" ").map(_.toDouble)).map(line=>Vectors.dense(line))
val rm=new RowMatrix(value)
rm.computeSVD(3)
1
2
3
4

主成分分析-降维二

对数据进项旋转变换的统计学方法===》线性空间进行及变化，然后投影在新坐标轴上的方差最大化==》新坐标轴为主成分

可以在较低维度的子空间尽可能表示原有数据性质。

1. 通过model

   用于原始数据是LabelPoint类型的情况，只需要RDD[Vector]的feature

   对其pca后放回

   import org.apache.spark.mllib.feature.PCA

   val pca=new PCA(3).fit(data.map(_.features)) //主成分个数为3

//构建PCA类，设定主成分个数为3，并用fit生成model

pca.

2. SVD分解来进行PVA变换

   val pc=rm.computePrincipalComponents(3)

   rm.multiply(pc)

3. 计算协方差矩阵

import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.linalg.distributed.RowMatrix
val data=sc.textFile("a.mat").map(_.split(" ")).map(_.toDouble).map(line=>Vectors.dense(line))
val rm=new RowMatrix(data)
val pc=rm.computePrincipalComponents(3)  //找前3个主成分
rm.multiply(pc)  //用矩阵乘法对原矩阵进行pca变换 从(4,9)==>(4,3)
1
2
3
4
5
6

import org.apache.spark.mllib.feature.PCA
import org.apache.spark.mllib.regression.LabeledPoint
val data=sc.textFile("a.mat").map(_.split(" ").map(_.toDouble)).map(line=>{
    LabelPoint(if(line(0)>1.0) 1.toDouble else 0.toDouble,Vectors.dense(line))
})
data.foreach(println)  //创建LabelPoint类型的数据
val pca=new PCA(3).fit(data.map(_.features))  //主成分个数为3，pca为对应主成分矩阵
val projected=data.map(p=>p.copy(features=pca.transform(p.features)))  //features成员替换为pca变换后的
projected.foreach(println)
1
2
3
4
5
6
7
8
9

分类算法

根据数据集的特点构造一个分类函数或分类模型(分类器)，把未知样本映射到给定类别中

f(x)=c x为特征向量 c为类别

1. 训练集T 创造模型
2. 模型带入算法
3. 测试集测试模型，看差分

二分-Logistic回归

来源

用于预测连续随机变量

x服从logistic分布：具有线性回归+sigmoid函数

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-efQnqBdt-1663928932140)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220922102242956.png)]

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Gk2tLPdi-1663928932141)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220922102308404.png)]

x是输入 y(0,1)输出 w为权重 b为偏置 w*x为w和x的内积

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Micwtpit-1663928932143)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220923172915989.png)]

极大相似法预测

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-BG3Ug2YZ-1663928932144)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220923172804064.png)]

对数相似法进行评估

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-m8xWmb3S-1663928932146)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220923172811768.png)]

new LogisticRegressionWithLBFGS()

LBFGS 表示用极大值求解非线性优化问题 Limited-memory BFGS 发明者BFGS

import org.apache.spark.mllib.classification.LogisticRegressionWithLBEGS
import org.apache.spark.mllib.evaluation.MulticlassMetrics
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.regression.LabeledPint
import org.aapche.spark.{SparkContext,SparkConf}

// 1.准备数据LabelPint

val data=sc.textFile("iris.data").map(p=>{
    val parts=p.split(",")
    LebeledPoint(if(parts(4)=="Iris-setosa") 0.toDouble 
                 else if(parts(4)=="Iris-versicolor") 1.toDouble)
    		else 2.toDouble,Vectors.dense(
            parts(0).toDouble,parts(1).toDouble,parts(2).toDouble,parts(3).toDouble,)
})

//数据分成训练集和测试机
val splits=data.randomSplit(Array(0.6,0.4))
val training=splits(0).cache()
val test=splits(1)
//val (train,test)=(splits(0),splits(1))
 //创建逻辑回归预测模型
val model=new LogisticRegressionWithLBFGS().setNumClasses(3).run(training)
//预测predict(feature)
val predata=test.map{
    case LabeledPoint(label,feature)=>{
        val prediction=model.predict(feature)
        (prediction,label)  //得到分类预测结果
    }
}
//评估
val metrics=new MulticlassMetrics(predata) //结果保存在MulticlassMetrics中
println(metrics.precision) //评估结果越接近1越好



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

决策树

一种基本的分类与回归方法，呈现树形结构

• 内部节点表示 一个属性测试

• 每个分支代表 一个测试输出

• 每个叶节点表 一种类别

根据损失函数最小化原则建立

1 特征选择(降维主成分分析/奇异值,特征类少就不用),可以用熵权法找

​ 熵是信息论的度量标准，熵越大不确定越大

​ 信息增益表示得知某一特征后使得信息不确定性减少的程度

信息增益

特征A对训练集D的信息增益g(D,A)==（D的熵-特征A给定D的熵）的差

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-3bw8uNp5-1663928932147)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220922115804439.png)]

信息增益比

信息增益 除以 训练集D关于特征A的值的熵之比

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-pkFkmqGZ-1663928932149)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220922120059682.png)]

基尼指数

1-样本属于第k类的概率的平方再相加

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-FoNwMY2Q-1663928932150)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220922120150926.png)]

2 决策树生成 org.apache.spark.mllib.tree.DecisionTree

​ DecisionTree.trainClassifier(训练集,分类数,分类特征信息,杂质,树最大深度,单个类最多几个值)

2.1.从根开始计算所有特征的信息增益选最大的作为节点特征

2.2.对子节点同样方法，构建决策树，直到没有特征或信息增均很小为止

2.3.提前停止:限定叶节点包含的最低数据量，可防止过度生长造成的过拟合

3 决策树剪枝

对已经生成的决策树进行简化

用极小化决策树整体的损失函数

参数a时的子树T的整体损失==预测误差+a*子树叶节点个数

a大：最优子树Ta小

a小：最优子树Ta大

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-AoNl0J9S-1663928932151)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220922120535741.png)]

import org.apache.spark.mllib.tree.DecisionTree
import org.apache.spark.mllib.regression.LabeledPoint
//获取数据，LabelPoint型的测试机test和训练集training   （data.randomSplit(Array(0.6,0.4)）
//获取模型
val model=Decision.trainClassifier(training,numClasses=3,Map[Int,Int](),impurify='gini',
                         maxDepth=3,maxBins=32)
//预测
val pre=test.map{point=>
    val prediction=model.predict(point.features)
    (point.label,prediction)
}
//输出树信息
println(model.toDebugString)
//评估 1-不一样数/总数  或者  一样数/总数
1- pre.filter(r=>r._1 != r._2).count.toDouble  /test.count()
//pre.filter(r=>r._1 == r._2).count.toDouble  /test.count()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

二分-SVM支持向量机

数之道 支持向量机SVM是什么，八分钟直觉理解其本质

分为：线性可分支持向量机 线性支持向量机 非线性支持向量机

训练数据线性可分：通过硬间隔最大化–线性可分支持向量机

训练数据近似线性可分：通过软间隔最大化—线性支持向量机(支持L1和L2正则化变形)

训练数据线性不可分：核技巧和软间隔最大化—非线性支持向量机

目的：不仅要将两类分开，而且要分类间隔足够大

红色的点为支持向量Support Vector Machine(SVM)

超平面描述

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ALB3iYDp-1663928932154)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220922150350944.png)]

损失函数：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-BGfSM7aH-1663928932156)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220922150410438.png)]

线性SVM用L2正则化训练，也支持L1正则化

L1和L2正则化”直观理解(之一)，从拉格朗日乘数法角度进行理解

什么是 L1 L2 正规化 正则化 Regularization (深度学习 deep learning)

//1.数据为LabelPoint的两个数据
//2.model=SVMWithSGD.train(训练数据,迭代次数(随机梯度下降算法))
//3.model.predict(point.features)
//4.设置阈值 大于阈值为正预测 下于阈值为负预测 ，model.setThreshold(0.0)
//5.评估矩阵BinaryClassificationMetrics  .areaUnderROC



//L1正则
//model=SVMSGD()
//model.optimizer.setNUmIterations(迭次数).setRegParam(浮点型正则化参数).setUpdater(new L1Updater) .run(训练集)=====》配置为L1

val data=sc.textFile("iris.data")
val parsedData=data.map{
    line=>{
        val parts=line.split(",")
        LabeledPoint(if(parts(4)=="Iris-setosa") 0.toDouble
                    else if (parts(4)=="Iris-versicolor") 1.toDouble
                    else 2.toDouble,
                    Vectors.dense(parts(0).toDouble,parts(1).toDouble,parts(2).toDouble,parts(3).toDouble,))
    }
}
val splits=parsedData.filter{
    point=>point.label!=2
}.randomSplit(Array(0.6,0.4))
val trainging=splits(0)
val test=splits(1)
val modelL2=new SVMWithSGD.train(training,10000)  //numIterations迭代次数为1000 stepSize迭代步伐 regParam正则化控制参数  miniBatchFraction 参与迭代计算的样本比例 initialWeights  weight向量初始值
val svm=new SVMWithSGD
svm.optimizer.setNumIterations(2000).setRegParam(0.1).setUpdater(new L1Updater)
val modelL1=modelL1.run(training)

//评估
model.clearThreshold()  //消除默认阈值
val scoreAndLabel=test.map{
    point=>{
        val score=model.predict(point.features)
        (score,point.label)
    }
}
scoreAndLabel.foreach(println)

model.clearThreshold(0.0)  //设置阈值，大于为正预测 小于为负预测
val scoreAndLabel=test.map{
    point=>{
        val score=model.predict(point.features)
        (score,point.label)
    }
}
scoreAndLabel.foreach(println)

val metrics=new BinaryClassificationMetrics(scoreAndLabels)
val auROC=metrics.areaUnderROC()
println(auROC)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

协同过滤算法-ALS

【推荐算法】经典算法之协同过滤 UserCF & ItemCF

对一组兴趣相同的用户或项目进行的推荐

协同过滤算法分为基于用户的协同过滤算法和基于项目的协同过滤算法

mllib支持基于模型的协同过滤

用户和商品通过一小组隐语义因子进行表达，并且这些因子也用于预测缺失的元素

mllib实现了交替最小二乘法ALS来学习这些隐性语义因子

显性反馈：明确表示用户对物品的喜好的行为

隐形反馈：不能明确反应用户喜好的行为：页面游览/点击/购买/喜欢/分享

将用户商品矩阵作为显性—与所观察到的用户偏好强度关联—用这个模型将找到的隐语义因子来预估一个用户对一个商品的偏好

model.recommendProducts给特定的用户推荐商品以及model.recommendUsers来给特定商品推荐潜在用户。

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

val data = sc.textFile("test.data")
val ratings = data.map(_.split(',') match { case Array(user, item, rate)
=> Rating(user.toInt, item.toInt, rate.toDouble)})
ratings.foreach{x => println(x)}
val splits = ratings.randomSplit(Array(0.8, 0.2))
val training = splits(0)
val test = splits(1)
val model = ALS.train(training, 10, 10, 0.01)
val model1 = new ALS()
  .setRank(params.rank) //隐语义因子个数
  .setIterations(params.iterations)  //迭代次数，
  .setLambda(params.lambda)  //ALS正则化参数 ，lambda越小且iterations越大=》均方差越小优
  .setImplicitPrefs(params.implicitPrefs) //使用显示还是隐式反馈数据集的版本
  .setUserBlocks(params.numUserBlocks) 
  .setProductBlocks(params.numProductBlocks)
  .run(training)
val testUsersProducts = test.map { case Rating(user, product, rate) =>
            (user, product)
          }
val predictions =
          model.predict(testUsersProducts).map { case Rating(user, product, rate)
              =>((user, product), rate)
           }
val ratesAndPreds = test.map { case Rating(user, product, rate) =>
         ((user, product), rate)
           }.join(predictions)
ratesAndPreds.foreach(println)   //((3,1),(1.0(真实结果),-0.22756397347958202(预测结果)))  
val MSE = ratesAndPreds.map { case ((user, product), (r1, r2)) =>
          val err = (r1 - r2)
            err * err
      }.mean()   // 1.0950191019929887不理想，需要调整参数或增加数据集
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

KMeans聚类算法-无监督学习

【五分钟机器学习】物以类聚的Kmeans

SparkML机器学习之聚类K-Means、GMM、LDA

KNN是有监督学习(还包括各种分类器)，找最近的样本，k指k个最近值

Kmeans是无监督学习，找最近的簇(中心点) ，k指分成几个簇

LMeans++/|| :思想史尽可能使得处理聚类中心互相远离

聚类以相似性为基础，为了使得同一类对象有更高的相似度

1. 使用样本的不同特征属性，根据某一给定的相似度度量方式(如欧氏距离，曼哈顿距离 马氏距离 皮尔逊相关系数 余弦 Jaccard相似系数)找相似的样本

2. 根据距离将样本划分成不同的组

   所谓聚类问题，就是给定一个元素集合D，其中每个元素具有n个可观察属性，使用某种算法将D划分成k个子集，要求每个子集内部的元素之间相异度尽可能低，而不同子集的元素相异度尽可能高。其中每个子集叫做一个簇。

   聚类分为划分算法，层次算法，密度算法，图论聚类法，网格算法，模型算法等

   用于图像处理 精准营销 生物信息学等

   KMeans(二分/流式) 高斯混合模型

   Power Iteration Clustering(PIC) 隐狄利克雷分布(LDA)

   1. 根据给定的k值，选取k个样本点作为初始划分中心；

   2. 计算所有样本点到每一个划分中心的距离，并将所有样本点划分到距离最近的划分中心；

   3. 计算每个划分中样本点的平均值，将其作为新的中心； 循环进行2~3步直至达到最大迭代次数，或划分中心的变化小于某一预定义阈值

import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.clustering.{KMeans, KMeansModel}

val rawData = sc.textFile("iris.csv")
val trainingData = rawData.map(line => {Vectors.dense(line.split(",").filter(p => p.matches("\\d*(\\.?)\\d*")).map(_.toDouble))}).cache()  //只要浮点数
//分3类 迭代100次 运行5次
val model : KMeansModel = KMeans.train(trainingData, 3, 100, 5) 
 model.clusterCenters.foreach(  //获取的聚类中心
     center => {
       println("Clustering Center:"+center)
     })

trainingData.collect().foreach(
      sample => {
        val predictedCluster = model.predict(sample)  //确定样本属于哪一个类
        println(sample.toString + " belongs to cluster " + predictedCluster)
      })
val wssse = model.computeCost(trainingData) //得到集合内的误差平方和WSSSE(Within Set Sum of Squared Error) 度量聚类的有效性，最优的k值是K-WSSSE的拐点Elbow位置
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

spark.ml

机械学习的步骤为：

• 源数据ETL(抽取 转化 加载)
• 数据预处理
• 指标提取
• 模型训练
• 交叉验证
• 新数据预测

DataFrame:被ML Pipeline用来存储数据

Transformer:转换器 将DF转成另一个DF的算法

Estimator:评估器 实现了fit方法 接受DF并产生一个转换器,比如随机森林算法就是一个Estimator

Parameter:用来设置Transformer或Estimator参数

PipeLine:工作流或管道。将多个状态(转换和评估器)连接在一起，形成机器学习的工作流获得结果

val pipeline=new Pipeline().setStages(Array(stage1,stage2,stage3...))
1

Tokenizer和HashingTf为transformers转换 LogisticRegression为Estimator评估

Tokenizer.transform将文档拆分为单词

HashingTF将字列转为特征向量并添加新列到DF.transform

LogisticRegression是评估器

LogisticRegression.fit创建moodel然后model.transform生成PipeLineModel并传到下一阶段

每个阶段transform方法更新数据级并传到下一阶段

构建工作流(ML Pipelines)

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
import org.apache.spark.ml.{Pipeline, PipelineModel}
import org.apache.spark.ml.classification.LogisticRegression
import org.apache.spark.ml.feature.{HashingTF, Tokenizer}
import org.apache.spark.mllib.linalg.Vector
import org.apache.spark.sql.Row
val sqlContext = new SQLContext(sc)
import sqlContext.implicits._
 val training = sqlContext.createDataFrame(Seq(
            (0L, "a b c d e spark", 1.0),
            (1L, "b d", 0.0),
            (2L, "spark f g h", 1.0),
            (3L, "hadoop mapreduce", 0.0)
          )).toDF("id", "text", "label")
val tokenizer = new Tokenizer().
            setInputCol("text").
            setOutputCol("words")
val hashingTF = new HashingTF().
            setNumFeatures(1000).
            setInputCol(tokenizer.getOutputCol).
            setOutputCol("features")
val lr = new LogisticRegression().
            setMaxIter(10).
            setRegParam(0.01)
//分词器--df器--逻辑回归器
val pipeline = new Pipeline().
            setStages(Array(tokenizer, hashingTF, lr))
//fit创建model  ,进行不断调整直到和原有数据的label差不多
val model = pipeline.fit(training)
val test = sqlContext.createDataFrame(Seq(
            (4L, "spark i j k"),
            (5L, "l m n"),
            (6L, "spark a"),
            (7L, "apache hadoop")
          )).toDF("id", "text")
//model 按照原有步骤 进行test
model.transform(test).
            select("id", "text", "probability", "prediction").
            collect().
            foreach { case Row(id: Long, text: String, prob: Vector, prediction: Double) =>
              println(s"($id, $text) --> prob=$prob, prediction=$prediction")
            }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

特征抽取TF-IDF Feature Extractors

特征抽取==从原有数据中抽取特征

特征转化=特征的维度 特征的转化 特征的修改

特征选取=从大规模特征集中选取一个子集

词频TF-IDF

TF(t,d)是t在文档d中出现的二次数 DF(t,D)是包含词语的文档的个数

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-jMeDV0UT-1663928932160)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220923094712608.png)]

|D|为总文档数

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-XfGAjlbh-1663928932161)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220923094749751.png)]

import org.apache.spark.sql.SparkSession
import org.apache.spark.ml.feature.{HashingTF, IDF, Tokenizer}
val spark=SparkSession().builder().master("local[*]").appName("test").getOrCreate()
val sentenceData = spark.createDataFrame(Seq(
            (0, "I heard about Spark and I love Spark"),
            (0, "I wish Java could use case classes"),
            (1, "Logistic regression models are neat")
          )).toDF("label", "sentence")
val tokenizer=new Tokenizer().setInputCol("sentence").setOutputCol("words")
val wordsData=tokenizer.transform(sentenceData)
val hashingTF=new HashingTF()
.setInputCol(tokenizer.getOutputCol)
.setOutputCol("rawFeatures").setNumFeatures(2000)
val featureizedData=hashingTF.transform(wordsData)
val idf=new IDF().setInputCol("tawFeatures").setOutputCol(features)
val idModel=idf.fit(featureizedData)
val rescaledData=idModel.transform(featurizedData)
rescaledData.show()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

特征抽取-CountVectorizer(Estimator)

通过计数来将文档转换成向量

相当于词频统计，vocabSize设置前几名 minDF设置至少几句话

import org.apache.spark.sql.SparkSession
import org.apache.spark.ml.feature.{CountVectorizer,CountVectorizerModel}
val spark=SparkSession().builder().master("local[*]").appName("test").getOrCreate()
val df = spark.createDataFrame(Seq(
            (0, Array("a", "b", "c")),
            (1, Array("a", "b", "b", "c", "a"))
          )).toDF("id", "words")
//给参数建模型
val cvModel: CountVectorizerModel = new CountVectorizer().
            setInputCol("words").
            setOutputCol("features").  
            setVocabSize(3). //词汇表最大含量，也就是只把前3个频率最高的放到词汇表里
            setMinDF(2).  //设定词汇表中的词至少在两个文档中出现过
            fit(df)  //建造模型
cvModel.transform(df).select("features").foreach { println }
//建造模型第二种
val cvm = new CountVectorizerModel(Array("a", "b", "c")).
            setInputCol("words").
            setOutputCol("features")
cvm.transform(df).select("features").foreach { println }    
cvm.show(truncate=false) //不截断
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

特征提取-WordsVec(Estimator) word to vector

用于NLP自然语言领域中

词嵌入 Word Embedding ，可以计算每个单词在其定语料库环境下的分布式词向量

词向量表示：在一定程度上刻画每个单词的语义。如果词的语义相近，词向量就在向量空间互相接近，使得词语的向量化建模更加精确而且改变现有方法并提高鲁棒性(rubust强壮健壮耐用性)。

用于机器翻译，标注问题，实体识别等

CBOW:每个词的上下文窗口向量来预测中心词的词向量

Skip-gram:每个中心词来预测其上下文窗口词并根据预测结果修正中心词的词向量

训练目标：学习词表征向量分布

优化目标：给定中心词的词向量的情况下最大化近似

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-iLDrhJm6-1663928932164)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220923114953248.png)]

w1…wt：一系列词序列，wt是中心词

wt+j(jE[-k,k])是上下文窗口中的词。

mei一个上下文窗口词wi在给定中心词wj下的条件概率由类似Softmax函数(相当于Sigmmoid函数的高位扩展版的形式计算)

uw: 当前词的词向量

vw:当前上下文的词向量

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-PlPAktuZ-1663928932165)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220923115218882.png)]

Softmax过于复杂，使用HUffman树进行层次Softmax(hierachical Softmax)进行优化(log p(wi|wj)从O(V)下降到O(log(V)))

【python-gensim】word2vec | 词向量 | 理论讲解+代码 | 文本分析

原理：采用一系列代表文档的词语来训练word2vecmodoel，该模型将每个词语映射到一个固定大小的向量，word2vecmodel使用文档中每个词语的平均数来将文档转换为向量，然后这个向量作为预测的特征，来计算文档相似度等等

import org.apache.spark.sql.SparkSession
import org.apache.spark.ml.feature.Word2Vec
val spark=SparkSession().builder().master("local[*]").appName("test").getOrCreate()
 val documentDF = sqlContext.createDataFrame(Seq(
            "Hi I heard about Spark".split(" "),
            "I wish Java could use case classes".split(" "),
            "Logistic regression models are neat".split(" ")
          ).map(Tuple1.apply)).toDF("text")  //文档，包含三个词语序列
val word2Vec = new Word2Vec().
            setInputCol("text").
            setOutputCol("result").
            setVectorSize(3).  //特征向量维度
            setMinCount(0)  //只有当某个词出现的频率大于等于0时才会到词汇表里，否则忽略
val model = word2Vec.fit(documentDF)  //生成model
val result = model.transform(documentDF) //把文档转为特征向量df
result.select("result").take(3).foreach(println)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

特征转化

参考来源

特征转化为了让它成为有效的特征，主要有标准化 归一化 特征的离散化等

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-5wLfP9kv-1663928932166)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220923135647916.png)]

//Binarizer
//初始化Binarizer对象并进行设定：setThreshold是设置我们的阈值，InputCol是设置需要进行二值化的输入列，setOutputCol设置输出列
    val binarizer = new Binarizer().setThreshold(50.0).setInputCol("age").setOutputCol("binarized_feature")
    //transform方法将DataFrame二值化。
    val binarizerdf = binarizer.transform(df)
-------------------------------------------------------------------------------------------------
//Bucketizer
// 设定边界，分为5个年龄组：[0,20),[20,30),[30,40),[40,50),[50,正无穷)
    // 注：人的年龄当然不可能正无穷，我只是为了给大家演示正无穷PositiveInfinity的用法，负无穷是NegativeInfinity。
    val splits = Array(0, 20, 30, 40, 50, Double.PositiveInfinity)
    //初始化Bucketizer对象并进行设定：setSplits是设置我们的划分依据
    val bucketizer = new Bucketizer().setSplits(splits).setInputCol("age").setOutputCol("bucketizer_feature")
    //transform方法将DataFrame二值化。
    val bucketizerdf = bucketizer.transform(df)
-------------------------------------------------------------------------------------------------
//QuantileDiscretizer
 //和Bucketizer类似：将连续数值特征转换离散化。但这里不再自己定义splits（分类标准），而是定义分几箱就可以了。
    val quantile = new QuantileDiscretizer().setNumBuckets(5).setInputCol("age").setOutputCol("quantile_feature")
    //因为事先不知道分桶依据，所以要先fit,相当于对数据进行扫描一遍，取出分位数来，再transform进行转化。
    val quantiledf = quantile.fit(df).transform(df)
-------------------------------------------------------------------------------------------------
//Standard
//**标准化数据=(原数据-均值)/标准差**
//**均值为0，方差为1**
    //setWithMean是否减均值。setWithStd是否将数据除以标准差。这里就是没有减均值但有除以标准差
//如果特征非常稀疏，并且有大量的0（现实应用中很多特征都具有这个特点），Z-score 标准化的过程几乎就是一个除0的过程，结果不可预料。所以在训练模型之前，一定要对特征的数据分布进行探索，并考虑是否有必要将数据进行标准化。基于特征值的均值（mean）和标准差（standard deviation）进行数据的标准化。
    df.printSchema()
    val scaler =new StandardScaler().setInputCol("features").setOutputCol("scaledFeatures").setWithMean(false).setWithStd(true)
    // 计算均值方差等参数
    val scalermodel = scaler.fit(df)
    // 标准化
    scalermodel.transform(df).show()
//标准化后的变量值围绕0上下波动，大于0说明高于平均水平，小于0说明低于平均水平。
-------------------------------------------------------------------------------------------------
    val maxabs = new MaxAbsScaler().setInputCol("features").setOutputCol("maxabs_features")
    // fit作用是把所有值都扫描一遍，计算出最大最小值，比如1000的话那么absMax=1000。最后返回MaxAbsScalerModel
    val scalerModel = maxabs.fit(dataFrame)
    // 使用每个特征的最大值的绝对值将输入向量的特征值都缩放到[-1,1]范围内
    val scalerdf = scalerModel.transform(dataFrame)
    scalerdf.show
-------------------------------------------------------------------------------------------------
    val maxabs = new MinMaxScaler().setInputCol("features").setOutputCol("minmax_features")
    val scalerModel = maxabs.fit(dataFrame)
    // 将所有值都缩放到[0,1]范围内
    val scalerdf = scalerModel.transform(dataFrame)
    scalerdf.show
-------------------------------------------------------------------------------------------------
    //setP是指L1正则化还是L2正则化，比如1.0就是上面说到的L1正则化，计算如下：1/(1+0.5+1)
    val normalizer = new Normalizer().setInputCol("features").setOutputCol("normalizer_features").setP(1.0)
    normalizer.transform(dataFrame).show(truncate = false)
-------------------------------------------------------------------------------------------------
//N-Gram认为语言中每个单词只与其前面长度 N-1 的上下文有关。主要分为bigram和trigram，bigram假设下一个词的出现依赖它前面的一个词，trigram假设下一个词的出现依赖它前面的两个词。在SparkML中用NGram类实现，setN(2)为bigram，setN(3)为trigram。
val ngram2 = new NGram().setN(2).setInputCol("words").setOutputCol("ngrams")
    val ngram3 = new NGram().setN(3).setInputCol("words").setOutputCol("ngrams")
    ngram2.transform(wordDataFrame).show(false)
    ngram3.transform(wordDataFrame).show(false)
-------------------------------------------------------------------------------------------------
 //setDegree表示多项式最高次幂 比如1.0,5.0可以是 三次：1.0^3 5.0^3 1.0+5.0^2 二次：1.0^2+5.0 1.0^2 5.0^2 1.0+5.0 一次：1.0 5.0
    val po = new PolynomialExpansion().setDegree(3).setInputCol("features").setOutputCol("Polynomial_features")
    po.transform(dataFrame).show(truncate = false)
-------------------------------------------------------------------------------------------------
// Tokenization（分词器）是一个接受文本（通常是句子）输入，然后切分成单词。
    val tokenizer = new Tokenizer().setInputCol("sentence").setOutputCol("words")
    tokenizer.transform(dataFrame).show()
    // RegexTokenizer（正则化分词器）基于正则表达式匹配提供了更多高级的分词功能。将gaps参数设置为false,表明使用正则表达式匹配标记，而不是使用分隔符。
    // 此处\\d表示匹配数字。
    val regextokenizer = new RegexTokenizer().setInputCol("sentence").setOutputCol("words").setGaps(false).setPattern("\\d")
    regextokenizer.transform(dataFrame).show()
    //使用udf函数 统计单词个数
    val wordcount = udf { (word: Seq[String]) => word.length }
    tokenizer.transform(dataFrame).select("words").withColumn("wordcount", wordcount(col("words"))).show()

-------------------------------------------------------------------------------------------------
    //__THIS__代表输入数据的基础表 在这里就是指df
    val sqltransformer1 = new SQLTransformer().setStatement("select id from __THIS__")
    val sqltransformer2 = new SQLTransformer().setStatement("select sum(score)/3 as average from __THIS__")
    sqltransformer1.transform(df).show()
    sqltransformer2.transform(df).show()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

标签和索引的转化

把标签数据转为正树索引，计算结束再转回去

转换器：StringIndexer IndexToString OneHoneEncoder VectorIndex

StringIndexer

字符串-》索引

数值-》字符串-》索引

val df1 = sqlContext.createDataFrame(
     |       Seq((0, "a"), (1, "b"), (2, "c"), (3, "a"), (4, "a"), (5, "c"))
     |     ).toDF("id", "category")
val df2 = sqlContext.createDataFrame(
     |       Seq((0, "a"), (1, "b"), (2, "c"), (3, "a"), (4, "a"), (5, "d"))
     |     ).toDF("id", "category")
val indexer = new StringIndexer().
     |       setInputCol("category").
     |       setOutputCol("categoryIndex")
val indexed1 = indexer.fit(df1).transform(df1)
val indexed2 = indexer.fit(df1).setHandleInvalid("skip").transform(df2) //StringIndexer构造df1但df2多一个d，用setHandleInvalid("skip")忽略d
val indexed3 = indexer.fit(df1)transform(df2)
1
2
3
4
5
6
7
8
9
10
11
12

IndexToString

索引-》字符型标签

一般都是和StringIndexer配合，先用StringIndexer转化成标签索引，进行模型训练，然后在预测标签的时候再把标签索引转化成原有的字符标签。(也可以使用自己提供的标签)

val df = sqlContext.createDataFrame(Seq(
     |       (0, "a"),
     |       (1, "b"),
     |       (2, "c"),
     |       (3, "a"),
     |       (4, "a"),
     |       (5, "c")
     |     )).toDF("id", "category")
val indexer = new StringIndexer().
     |       setInputCol("category").
     |       setOutputCol("categoryIndex").
     |       fit(df)   //字符标签转为索引到categoryIndex列上
 val indexed = indexer.transform(df)   //运行StringIndexer到df上 得到indexed结果
val converter = new IndexToString().
     |       setInputCol("categoryIndex").
     |       setOutputCol("originalCategory")  //把索引输出到orginalCategory列上
val converted = converter.transform(indexed)  //运行IndexToString到df上 得到converted
 converted.select("id", "originalCategory").show()  //输出converted
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

OneHotEncoder

标签索引-》二进制数组

适合期望类别特征为连续特征的算法，比如Logistic regression逻辑回归

val df = sqlContext.createDataFrame(Seq(
     |       (0, "a"),
     |       (1, "b"),
     |       (2, "c"),
     |       (3, "a"),
     |       (4, "a"),
     |       (5, "c"),
     |       (6, "d"),
     |       (7, "d"),
     |       (8, "d"),
     |       (9, "d"),
     |       (10, "e"),
     |       (11, "e"),
     |       (12, "e"),
     |       (13, "e"),
     |       (14, "e")
     |     )).toDF("id", "category")
val indexer = new StringIndexer().
     |       setInputCol("category").
     |       setOutputCol("categoryIndex").
     |       fit(df)
val indexed = indexer.transform(df)
 val encoder = new OneHotEncoder().
     |       setInputCol("categoryIndex").
     |       setOutputCol("categoryVec")
val encoded = encoder.transform(indexed)
 encoded.show()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

VectorIndexer

类别特征索引，自动识别哪些特则会那个是类别型的，并将原始值转为类别索引

1. 获得一个向量类型的输入以及maxCategories参数。

2. 基于不同特征值的数量来识别哪些特征需要被类别化，其中最多maxCategories个特征需要被类别化。

3. 对于每一个类别特征计算0-based（从0开始）类别索引。

4. 对类别特征进行索引然后将原始特征值转换为索引。

索引后的类别特征可以帮助决策树等算法恰当的处理类别型特征，并得到较好结果。

val data = Seq(Vectors.dense(-1.0, 1.0, 1.0),Vectors.dense(-1.0, 3.0, 1.0), Vectors.dense(0.0, 5.0, 1.0))
val df = spark.createDataFrame(data.map(Tuple1.apply)).toDF("features")
val indexer = new VectorIndexer().
     |       setInputCol("features").
     |       setOutputCol("indexed").
     |       setMaxCategories(2)  //只有类别小于2的特征才被认为是类别型特征，否则连续型特征，0则为编号0
val indexerModel = indexer.fit(df)
val categoricalFeatures: Set[Int] = indexerModel.categoryMaps.keys.toSet
println(s"Chose ${categoricalFeatures.size} categorical features: " + categoricalFeatures.mkString(", "))
val indexedData = indexerModel.transform(df)
indexedData.foreach { println }
1
2
3
4
5
6
7
8
9
10
11

Logistic回归分类器

【机器学习】逻辑回归十分钟学会，通俗易懂（内含spark求解过程）

逻辑回归是统计学习的经典分类方法，属于对数线性模型

因变量可以是二分类的，也可以是多分类的

数据挖掘的算法，解决二分类问题的

判断：每个数据属于哪一类

二分类：类型只有两种

回归输出连续 分类输出离散

如图：左回归右分类

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-aezu4TyZ-1663928932167)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220923170052663.png)]

逻辑回归=线性回归+sigmoid函数

线性回归：一条线，变量y和结果y的关系y=ax+b

把回归分类：y>N为1赚钱 y<N为0亏钱

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-AyCJaJB2-1663928932169)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220923170336271.png)]

x=0:y=0.5

x<0 :y=0

x>0:y=1

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-v9oNFka0-1663928932171)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220923170624988.png)]

把线性函数z=kx+b结果z当成因变量放到sigmoid函数1/1+e^(-z)里面

逻辑回归

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-uimt8ciA-1663928932172)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220923170844353.png)]

用损失函数得到预测值和实际值相似程度，C越小相似越高，模型越好！

如何通俗地解释梯度下降法

梯度下降法：

快速找到最低点的方法

问题：方向+距离+终止条件是什么

梯度方向：梯度增加：函数值增加的方向

梯度反方向：梯度减少：函数值减少的方向

梯度=f’ 函数f(x)的导函数 x1=x0-f(x0)’ =-10（x0=10 f(x)'=2x) =》没有距离会来回震荡

距离/学习率：需要选择合适的学习率，查看梯度的值和下降的速度

终值值：梯度值作为终止条件(例如0，0.001))

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-wqtb425q-1663928932173)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220923172401410.png)]

import org.apache.spark.ml.{Pipeline,PipelineModel}
import org.apache.spark.ml.classification.LogisticRegression
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.{IndexToString, StringIndexer, VectorIndexer,HashingTF, Tokenizer}
import org.apache.spark.mllib.linalg.{Vector,Vectors}
import org.apache.spark.sql.Row
import org.apache.spark.mllib.stat.{MultivariateStatisticalSummary, Statistics}
import org.apache.spark.ml.classification.LogisticRegressionModel

val observations=sc.textFile("G:/spark/iris.data").map(_.split(",")).map(p => Vectors.dense(p(2).toDouble, p(3).toDouble))
 val summary: MultivariateStatisticalSummary = Statistics.colStats(observations)
println(summary.mean/variance/numNonzeros)
case class Iris(features: Vector, label: String)
 val data = sc.textFile("G:/spark/iris.data")
          |       .map(_.split(","))
          |       .map(p => Iris(Vectors.dense(p(2).toDouble, p(3).toDouble), p(4).toString()))
          |       .toDF()
data.show()
data.registerTempTable("iris")  //临时表
val df = sqlContext.sql("select * from iris where label != 'Iris-setosa'")
df.map(t => t(1)+":"+t(0)).collect().foreach(println)  //失去了Iris-setosa
val labelIndexer = new StringIndexer().setInputCol("label").setOutputCol("indexedLabel").fit(df) //字符串索引
val featureIndexer = new VectorIndexer().setInputCol("features").setOutputCol("indexedFeatures").fit(df)//类别索引
val Array(trainingData, testData) = df.randomSplit(Array(0.7, 0.3))
val lr = new LogisticRegression().   //逻辑回归库 通过explainParams()获取参数设置
     |       setLabelCol("indexedLabel").  
     |       setFeaturesCol("indexedFeatures").
     |       setMaxIter(10).  //循环10次
     |       setRegParam(0.3).  //正则化项
     |       setElasticNetParam(0.8)
println("LogisticRegression parameters:\n" + lr.explainParams() + "\n")


val labelConverter = new IndexToString().
     |       setInputCol("prediction").
     |       setOutputCol("predictedLabel").
     |       setLabels(labelIndexer.labels)
 val pipeline = new Pipeline().
     |       setStages(Array(labelIndexer, featureIndexer, lr, labelConverter))
val model = pipeline.fit(trainingData)
val predictions = model.transform(testData) //预测
predictions.
     |       select("predictedLabel", "label", "features", "probability").
     |       collect().
     |       foreach { case Row(predictedLabel: String, label: String, features: Vector, prob: Vector) =>
     |         println(s"($label, $features) --> prob=$prob, predictedLabel=$predictedLabel")
     |     }
val evaluator = new MulticlassClassificationEvaluator().
     |       setLabelCol("indexedLabel").
     |       setPredictionCol("prediction")//评估
val accuracy = evaluator.evaluate(predictions)
val lrModel = model.stages(2).asInstanceOf[LogisticRegressionModel]  //用stages(2)获取lr模型
println("Coefficients: " + lrModel.coefficients+"Intercept: "+lrModel.intercept+
     |         "numClasses: "+lrModel.numClasses+"numFeatures: "+lrModel.numFeatures)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

决策树

什么是决策树（Decision Tree）？【知多少】

为了目的而根据条件进行选择的过程

通常用于分类 ，由节点和边构成

最初节点：根节点 分支节点:子节点 没有分支的节点：叶子节点

叶子节点代表样本的分类结果

边代表方向

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-k6YygWgC-1663928932175)(C:\Users\cry\AppData\Roaming\Typora\typora-user-images\image-20220923174037455.png)]

根据熵(系统内部的混乱程度)来代表分支下样本种类的丰富性

分支下样本种类越多越混乱熵越多，分支下样本属于同一类熵就为0

构建决策树：

随着树的深度(层数的增加)让熵快速降低，熵降低的越快代表分类效率越高

优点：具有天然可解释性

缺点：完美分类一定过拟合

需要去掉一些分支-剪枝：

1. 预剪枝：开始前规定一些条件，比如树的深度(层数)达到某一条件就结束
2. 后剪枝：先建立树，再依据条件限制叶子节点(无分支的节点)个数，去点一部分分支

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
import org.apache.spark.ml.{Pipeline,PipelineModel}
import org.apache.spark.ml.classification.DecisionTreeClassificationModel
import org.apache.spark.ml.classification.DecisionTreeClassifier
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.{IndexToString, StringIndexer, VectorIndexer,HashingTF, Tokenizer}
import org.apache.spark.mllib.linalg.{Vector,Vectors}
import org.apache.spark.sql.Row
import org.apache.spark.mllib.stat.{MultivariateStatisticalSummary, Statistics}
val observations=sc.textFile("G:/spark/iris.data").map(_.split(",")).map(p => Vectors.dense(p(2).toDouble, p(3).toDouble))
case class Iris(features: Vector, label: String)
val data = sc.textFile("G:/spark/iris.data")
          |       .map(_.split(","))
          |       .map(p => Iris(Vectors.dense(p(2).toDouble, p(3).toDouble), p(4).toString()))
          |       .toDF()
data.show()
val labelIndexer = new StringIndexer().setInputCol("label").setOutputCol("indexedLabel").fit(data)
 val featureIndexer = new VectorIndexer().setInputCol("features").setOutputCol("indexedFeatures").fit(data)
val Array(trainingData, testData) = data.randomSplit(Array(0.7, 0.3))
 val dt = new DecisionTreeClassifier()
  |            .setLabelCol("indexedLabel")
  |            .setFeaturesCol("indexedFeatures")
  |            .setImpurity("gini")
  |            .setMaxDepth(5)
println("DecisionTreeClassifier parameters:\n" + dt.explainParams() + "\n")
val labelConverter = new IndexToString().
     |       setInputCol("prediction").
     |       setOutputCol("predictedLabel").
     |       setLabels(labelIndexer.labels)
 val pipeline = new Pipeline().
     |       setStages(Array(labelIndexer, featureIndexer, dt, labelConverter))
 val model = pipeline.fit(trainingData)
 val predictions = model.transform(testData)
predictions.
     |       select("predictedLabel", "label", "features").
     |       collect().
     |       foreach { case Row(predictedLabel: String, label: String, features:
 Vector) =>
     |         println(s"($label, $features) --> predictedLabel=$predictedLabel")
     |     }
val evaluator = new MulticlassClassificationEvaluator().
     |       setLabelCol("indexedLabel").
     |       setPredictionCol("prediction")
val accuracy = evaluator.evaluate(predictions)
val treeModel = model.stages(2).asInstanceOf[DecisionTreeClassificationModel]
println("Learned classification tree model:\n" + treeModel.toDebugString)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

OutputCol(“indexedLabel”).fit(data)
val featureIndexer = new VectorIndexer().setInputCol(“features”).setOutputCol(“indexedFeatures”).fit(data)
val Array(trainingData, testData) = data.randomSplit(Array(0.7, 0.3))
val dt = new DecisionTreeClassifier()
| .setLabelCol(“indexedLabel”)
| .setFeaturesCol(“indexedFeatures”)
| .setImpurity(“gini”)
| .setMaxDepth(5)
println(“DecisionTreeClassifier parameters:\n” + dt.explainParams() + “\n”)
val labelConverter = new IndexToString().
| setInputCol(“prediction”).
| setOutputCol(“predictedLabel”).
| setLabels(labelIndexer.labels)
val pipeline = new Pipeline().
| setStages(Array(labelIndexer, featureIndexer, dt, labelConverter))
val model = pipeline.fit(trainingData)
val predictions = model.transform(testData)
predictions.
| select(“predictedLabel”, “label”, “features”).
| collect().
| foreach { case Row(predictedLabel: String, label: String, features:
Vector) =>
| println(s"($label, $features)-->predictedLabel=$predictedLabel")
| }
val evaluator = new MulticlassClassificationEvaluator().
| setLabelCol(“indexedLabel”).
| setPredictionCol(“prediction”)
val accuracy = evaluator.evaluate(predictions)
val treeModel = model.stages(2).asInstanceOf[DecisionTreeClassificationModel]
println(“Learned classification tree model:\n” + treeModel.toDebugString)

1