Machine Learning on Spark -- Basic Statistics

1.Summary statistics 统计量

colStats提供了针对向量形式RDD 的列的统计量。colStats() 返回了一个MultivariateStatisticalSummary的实例, which contains the column-wise max, min, mean, variance, and number of nonzeros, as well as the total count.

//导入包
import org.apache.spark.mllib.linalg.vector
import org.apache.apark.mllib.stat.{MultivariateStatisticalSummary,Statistics}

//创建向量
val observations:RDD[Vector] = sc.parallelize(Array(Array(1,2,3),Array(2,3,4),Array(3,4,5))).map(d=>vector.dense(d))

// 计算向量的列的统计量
val summary:MultivariatesStatisticalSummary = Statistics.colstats(oberservations)

//基本统计量
println(summary.mean)
println(summary.variance)   //计算标准差
println(summary.numNonzeros)  //计算非0个数
println(summary.max)
println(summary.min)

//距离计算
//计算并打印欧几里得距离
println(summary.normL2)

//计算并打印曼哈顿距离
println(summary.normL1)

2. Correlation 相关性

计算两个序列数据的相关性。在spark.mllib中提供给了非常灵活的方式在多个序列中计算两两相关性。目前支持的相关性方法为皮尔森相关性与斯皮曼相关性。

在计算相关性之后，会输出一个相关性矩阵或者一个双精度数值。

具体的API可以参见statistics Scala docs

import org.apache.spark.sparkConf
import org.apache.spark.sparkContext
import org.apache.spark.mllib.linalg._
import org.apache.spark.mllib.stat.statistics
import org.apache.spark.rdd.RDD

创建一个对象
object Correlation {

//设置环境变量
val sparkConf = new sparkConf()
.setAppName(Correlation)
.setMaster("Local")
val sparkContext = new sparkContext(sparkConf)

//创建两个序列series,两个序列的数据数目必须一样
val series1:RDD[Double]=sc.parallelize(Array(10,11,14,16,18))
val series2:RDD[Double]=sc.parallelize(Array(9,10,13,16,17))

//计算pearson相关系数(默认为pearson法)
val correlation1:Double = Statistics.corr(series1,series2,"pearson")

//计算spearman相关系数
val correlation2:Double = Statistics.corr(series1,series2,"pearman")

//创建一个矩阵RDD[Vector]
val data:RDD[Vector] = sc.parallelize((Array(1,2,3),Array(2,3,4),Array(3,4,5))).map(d=>vector.dense(d))

//计算pearson相关性，默认使用
val correlation3:Matrix = Statistics.corr(data,"pearson")
}

注：不同相关系数代表的意义不同。pearson相关系数代表两组数据的余弦分开程度，随着数据量增加，两组数据差别将增大；spearman相关系数更注重两组数据的拟合度，两组数据随数据量增加增长曲线不变。

3.Stratified sampling 分层抽样

分层抽样是一种常用的抽样方法，先将总体的单位按某种特征分成若干不交叉的层级，再从每个层级中随机抽样，组成一个样本。

分层抽样的数据使用map进行标记，故格式为(key,value),key相当于类别标签，可以是男与女的性别类别，也可以是文件的ID，或年龄层次的分类等等。

以spark.readme文件为例

import org.apache.spark.sparkConf
import org.apache.spark.sparkContext
import org.apache,spark.sparkContext._
import org.apache.spark.rdd.PairRDDFunctions

object Sampling{
def main(args:Aarray[String]){
//设置环境变量
val sparkConf = new sparkConf()
.setAppName("Samping")
.setMaster("")
val sparkContext = new sparkContext(sparkConf)

//导入文件，创建实例
val data = sc.textFile("/README.md")

//进行wordcount运算，形成(key,value)形式
val wordcount = data.flatmap(line=>line.split(" ")).map(word=>(word,1).reduceByKey((a,b)=>a+b)

//定义key为spark，抽样比例为50%
val fractions:Map[String,Double] = Map("spark"=>0.5)

//抽样方式一:sampleByKey(withReplacement = false, fractions)

//抽样方式二:samlebyKeyExact(withReplacement = false,fractions)
}
}

注：使用sampleByKeyExact方法进行采样，该方法资源消耗较sampleByKey更大，但采样后的大小与预期大小更接近，可信度达到99.99%

4.Hypothesis testing 假设检验

目前spark.mllib支持pearson’s chi-sqaured test,中文叫卡方检验。

其中拟合优度卡方检验输入的是向量vector,独立样本的卡方检验输入的是矩阵matrix.

spark.mllib也支持输入RDD[LabeledPoint],通过卡方独立性检验来筛选特征。

import org.apache.spark.mllib.linalg.{Matrices,Vectors}
import org.apache.spark.mllib.stat.Statistics

object ChiTest{
def main(args:Aarray[String]){

//对单向量进行卡方检验
val dv = Vectors.dense(1,2,3,4,5,6)
val dvChi =  Statistics.chiSqTest(dv)

//对矩阵数据组进行卡方检验
val mat = Mattrices.dense(3,2,Array(1,2,3,4,5,6))
val matChi = Statistics.chiSqTest(mat)
}

计算后会打印出三个结果数据：

自由度，统计量，P值

若P值<0.05则通常表示不存在显著差异

5.Random data generation 生成随机数

给定分布（如uniform,标准正态分布，泊松分布..)，spark.mllib 支持产生随机RDD

RandomRDDs 提供了factory method 来产生随机的双精度数值RDD,或者向量RDD。

import org.apache.sparkContext
import org.apache.spark.mllib.random.RandomRDD

//生成100个服从标准正态分布（0,1）的随机数，分区数设为10
val u = normalRDD(sc,100L,10)

//将上面的U转换成服从（1,4）的正态分布
val v = u.map(x=>1.0+2.0*x)

6.Kernel density estimation 核密度估计

核密度估计是一种可视化经验概率分布的有效技术，它无需基于给定样本的概率分布。给出一系列的点，它可以计算出概率密度函数的估计量。

import org.apache.spark.mllib.stat.kernelDensity
import org.spache.spark.rdd.RDD

val data: RDD[Double] = ... // an RDD of sample data

// Construct the density estimator with the sample data and a standard deviation for the Gaussian
// kernels
val kd = new KernelDensity()
.setSample(data)
.setBandwidth(3.0)

// Find density estimates for the given values
val densities = kd.estimate(Array(-1.0, 2.0, 5.0))