spark机器学习从0到1基本的统计工具之(三）

给定一个数据集，数据分析师一般会先观察一下数据集的基本情况，称之为汇总统计或者概要性统计。一般的概要性统计用于概括一系列观测值，包括位置或集中趋势（比如算术平均值、中位数、众数和四分位均值），展型（比如四分位间距、绝对偏差和绝对距离偏差、各阶矩等），统计离差，分布的形状，依赖性等。除此之外，spark.mllib库也提供了一些其他的基本的统计分析工具，包括相关性、分层抽样、假设检验，随机数生成等。

一、概括统计 summary statistics

我们通过统计学中提供的函数colStats为RDD [Vector]提供列摘要统计信息。我们可以获得每一列的最大值，最小值，均值、方差、总数。

我们用UCI 提供的莺尾花的数据来举例。数据下载地址：http://archive.ics.uci.edu/ml/machine-learning-databases/iris/。我们将鸢尾花的四个属性，即萼片长度，萼片宽度，花瓣长度和花瓣宽度存储在observations中，类型为RDD[Vector]。

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaSparkContext;

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.mllib.linalg.Vector;

import org.apache.spark.mllib.linalg.Vectors;

import  org.apache.spark.mllib.stat.MultivariateStatisticalSummary;

import org.apache.spark.mllib.stat.Statistics;

SparkConf conf = new  SparkConf().setAppName("colStatsTest").setMaster("local");

JavaSparkContext  sc = new  JavaSparkContext(conf);

JavaRDD<String> source =  sc.textFile("data/iris.data"); // 读取数据

JavaRDD<Vector> observations = source.map(line  -> {

        String[] parts = line.split(",");

        return  Vectors.dense(Double.valueOf(parts[0]), Double.valueOf(parts[1]),

Double.valueOf(parts[2]), Double.valueOf(parts[3]));//将RDD<String>转化为RDD<Vector>

});

MultivariateStatisticalSummary summary =  Statistics.colStats(observations.rdd()); //计算列摘要统计信息。

System.out.println(summary.count()); //总数(long)

System.out.println(summary.mean());  //  包含每列平均值的密集向量(vector)

System.out.println(summary.variance());   // 列方差(vector)

System.out.println(summary.max());   // 最大值(vector)

System.out.println(summary.min());   // 最小值(vector)

System.out.println(summary.normL1());   //每列的L1范数(vector)

System.out.println(summary.normL2());   //每列的L2范数(vector)

System.out.println(summary.numNonzeros()); //  每列中的非零数(vector)

二、相关性

计算两个数据系列之间的相关性是统计学中的常见操作。在spark.mllib中，我们提供了计算多个系列之间成对相关性的灵活性。支持的相关方法目前是Pearson和Spearman的相关性。

相关系数是用以反映变量之间相关关系密切程度的统计指标。简单的来说就是相关系数绝对值越大（值越接近1或者-1）,当取值为0表示不相关，取值为(0~-1]表示负相关，取值为(0, 1]表示正相关。

2.1、Pearson相关系数

Pearson相关系数表达的是两个数值变量的线性相关性, 它一般适用于正态分布。其取值范围是[-1, 1], 当取值为0表示不相关，取值为[-1~0)表示负相关，取值为(0, 1]表示正相关。

Pearson相关系数

2.2、Spearman相关系数

Spearman相关系数也用来表达两个变量的相关性，但是它没有Pearson相关系数对变量的分布要求那么严格，另外Spearman相关系数可以更好地用于测度变量的排序关系。其计算公式为：

Spearman相关系数

统计提供了计算序列之间相关性的方法。根据输入类型，两个JavaDoubleRDD或JavaRDD <Vector>，输出将分别为Double或相关矩阵。

JavaDoubleRDD seriesX =  source.mapToDouble(x->Double.parseDouble(x.split(",")[0]));

JavaDoubleRDD seriesY =  source.mapToDouble(x->Double.parseDouble(x.split(",")[1])); //必须具有与seriesX相同数量的分区和基数

Double correlation =  Statistics.corr(seriesX.srdd(), seriesY.srdd(),  "pearson");//使用Pearson方法计算相关性。为Spearman的方法输入“spearman”。如果未指定方法，默认情况下将使用Pearson的方法。

System.out.println("Correlation is: " +  correlation);

/**控制台输出结果：

----------------------------------------------------

Correlation is: -0.10936924995062468

-----------------------------------------------------

**/

说明数据集的前两列，即花萼长度和花萼宽度具有微小的负相关性。

JavaRDD<Vector> data = source.map(line -> {

    String[] parts = line.split(",");

    return  Vectors.dense(Double.valueOf(parts[0]),

    Double.valueOf(parts[1]));//将RDD<String>转化为RDD<Vector>

}); //请注意，每个Vector都是一行而不是一列

Matrix correlMatrix =  Statistics.corr(data.rdd(), "pearson");

System.out.println(correlMatrix.toString());

/**控制台输出结果：

-----------------------------------------------------

**1.0                   -0.10936924995062468

**-0.10936924995062468  1.0

-------------------------------------------------------

**/

三、分层抽样 Stratified sampling

与spark.mllib中的其他统计函数不同，可以对RDD的键值对执行分层抽样方法sampleByKey和sampleByKeyExact。对于分层抽样，可以将键视为标签，将值视为特定属性。例如，密钥可以是人或女人，或文档ID，并且相应的值可以是人口中的人的年龄列表或文档中的单词列表。 sampleByKey方法将翻转硬币以决定是否对样本进行采样，因此需要对数据进行一次传递，并提供预期的样本大小。 sampleByKeyExact比sampleByKey中使用的每层简单随机抽样需要更多的资源，但是会提供99.99％置信度的精确抽样大小。

3.1、sampleByKey 方法

sampleByKeyExact（）允许用户准确地采样⌈fk⋅nk⌉∀k∈K项，其中fk是密钥k的期望分数，nk是密钥k的键 - 值对的数量，K是密钥集。无需更换的采样需要在RDD上额外通过一次以确保样本大小，而更换采样则需要两次额外的通过。

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaSparkContext;

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.api.java.JavaPairRDD;

import org.spark_project.guava.collect.ImmutableMap;

import scala.Tuple2;

List<Tuple2<String, String>> list =  Arrays.asList(

new Tuple2<>("female","Lily"),

new Tuple2<>("female","Lucy"),

new Tuple2<>("female","Emily"),

new Tuple2<>("female","Kate"),

new Tuple2<>("female","Alice"),

new Tuple2<>("male","Tom"),

new Tuple2<>("male","Roy"),

new Tuple2<>("male","David"),

new Tuple2<>("male","Frank"),

new Tuple2<>("male","Jack"));//创建了一组数据，分成 “female” 和  “male” 两类

JavaPairRDD<String, String> data =  sc.parallelizePairs(list);

ImmutableMap<String, Double> fractions =  new ImmutableMap.Builder<String,Double>()

.put("female",0.6)

.put("male",0.4)

.build(); //从每个键Map <K，Double>中指定所需的精确分数

这里，设置采取60%的female和40%的male，因为数据中female和male各有5个样本，所以理想中的抽样结果应该是有3个female和2个male。接下来用sampleByKey进行抽样：

JavaPairRDD<String, String> approxSample = data.sampleByKey(false, fractions,1); //从每个层获取大致样本

approxSample.foreach(x->{

System.out.println(x); //打印approxSample分层取样的结果;

});

/** *控制台输出结果：

-----------------------

(female,Lily)

(female,Lucy)

(female,Emily)

(female,Kate)

(male,Roy)

(male,Frank)

--------------------

**/

从上面可以看到，本应该抽取3个female和2个male，但结果抽取了5个female和1个male，结果并不够准确，不过在样本数量足够大且要求一定效率的情况下，用sampleByKey进行抽样还是可行的。

3.2、sampleByKeyExact 方法

sampleByKey 和 sampleByKeyExact 的区别在于 sampleByKey 每次都通过给定的概率以一种类似于掷硬币的方式来决定这个观察值是否被放入样本，因此一遍就可以过滤完所有数据，最后得到一个近似大小的样本，但往往不够准确。而 sampleByKeyExtra 会对全量数据做采样计算。对于每个类别，其都会产生（fk⋅nk）个样本，其中fk是键为k的样本类别采样的比例；nk是键k所拥有的样本数。 sampleByKeyExtra 采样的结果会更准确，有99.99%的置信度，但耗费的计算资源也更多。

JavaPairRDD<String, String> exactSample = data.sampleByKeyExact(false, fractions,1);

exactSample.foreach(x->{

System.out.println(x);//打印exactSample分层取样的结果;});

/***控制台输出结果：

--------------------

(female,Lily)

(female,Emily)

(female,Kate)

(male,Roy)

(male,Frank)

--------------------

**/

从实验结果可以看出，所得结果和预想一致，但当样本数量比较大时，可能会耗时较久。

四、假设检验 hypothesis testing

假设检验是统计学中一种强有力的工具，用于确定结果是否具有统计显着性，无论该结果是否偶然发生。 spark.mllib目前支持Pearson的卡方（χ2）测试，以确保拟合和独立性。不同的输入类型决定了是做拟合度检验还是独立性检验。拟合度检验要求输入为Vector, 独立性检验要求输入是Matrix。

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.api.java.JavaSparkContext;

import org.apache.spark.mllib.linalg.Matrices;

import org.apache.spark.mllib.linalg.Matrix;

import org.apache.spark.mllib.linalg.Vector;

import org.apache.spark.mllib.linalg.Vectors;

import org.apache.spark.mllib.stat.Statistics;

import org.apache.spark.mllib.stat.test.ChiSqTestResult;

JavaRDD<Vector> vectors1 = source.map(x->{

String[] splits = x.split(",");

return  Vectors.dense(Double.parseDouble(splits[0]),

Double.parseDouble(splits[1]),

Double.parseDouble(splits[2]),

Double.parseDouble(splits[3]));

});

Vector v1 = vectors1.first();//获取  iris数据集中的第1条记录，类型为Vector

Vector v2 = vectors1.take(2).get(1);//获取 iris数据集中的第2条记录，类型为Vector

4.1、适合度检验 Goodness fo fit

ChiSqTestResult goodnessOfFitTestResult = Statistics.chiSqTest(v1);

System.out.println("goodnessOfFitTestResult:"+goodnessOfFitTestResult);

/***控制台输出结果：

------------------------------------------------------------------------------------------------------------------------

goodnessOfFitTestResult:Chi squared test summary:

method: pearson

degrees of freedom = 3

statistic = 5.588235294117647

pValue = 0.1334553914430291

No presumption against null hypothesis: observed follows the same distribution as expected..

-----------------------------------------------------------------------------------------------------------------------------

**/

可以看到P值，自由度，检验统计量，所使用的方法，以及零假设等信息。我们先简单介绍下每个输出的意义：

method: 方法。这里采用pearson方法。

statistic：检验统计量。简单来说就是用来决定是否可以拒绝原假设的证据。检验统计量的值是利用样本数据计算得到的，它代表了样本中的信息。检验统计量的绝对值越大，拒绝原假设的理由越充分，反之，不拒绝原假设的理由越充分。

degrees of freedom：自由度。表示可自由变动的样本观测值的数目，

pValue：统计学根据显著性检验方法所得到的P 值。一般以P < 0.05 为显著， P<0.01 为非常显著，其含义是样本间的差异由抽样误差所致的概率小于0.05 或0.01。

一般来说，假设检验主要看P值就够了。在本例中pValue =0.133，说明两组的差别无显著意义。通过V1的观测值[5.1, 3.5, 1.4, 0.2]，无法拒绝其服从于期望分配（这里默认是均匀分配）的假设。

4.2、独立性检验 Indenpendence

卡方独立性检验是用来检验两个属性间是否独立。其中一个属性做为行，另外一个做为列，通过貌似相关的关系考察其是否真实存在相关性。比如天气温变化和肺炎发病率。

Matrix mat = Matrices.dense(2, 2, new double[]{v1.apply(0), v1.apply(1),v2.apply(0), v2.apply(1)});

System.out.println("mat:"+mat);

/***

控制台输出结果：

---------------------------

mat:5.1 4.9 3.5 3.0

----------------------------

**/

同样的，键值对也可以进行独立性检验，这里我们取iris的数据组成键值对：

JavaRDD<LabeledPoint> LabeledPoints =  source.map(x->{

String[] splits = x.split(",");

Double label = 0.0;

if(splits[4].equals("Iris-setosa"))  {

label = 0.0;

}else  if(splits[4].equals("Iris-versicolor")) {

label = 1.0;

}else {

label = 2.0;

}

return new  LabeledPoint(label,Vectors.dense(Double.parseDouble(splits[0]),

Double.parseDouble(splits[1]),

Double.parseDouble(splits[2]),

     Double.parseDouble(splits[3])));

});

ChiSqTestResult[] c2 =  Statistics.chiSqTest(LabeledPoints);

for (ChiSqTestResult chiSqTestResult :  c2) {

System.out.println("c2:"+chiSqTestResult);}

/***控制台输出结果：

-----------------------------------------------------------------------------------------------------------------------------------------------------

c2:Chi squared test summary:

method: pearson

degrees of freedom = 68

statistic = 156.26666666666665

pValue = 6.6659873176888595E-9

Very strong presumption against null hypothesis: the occurrence of the outcomes is  statistically independent..

c2:Chi squared test summary:

method: pearson

degrees of freedom = 44

statistic = 88.36446886446883

pValue = 8.303947787857702E-5

Very strong presumption against null hypothesis: the occurrence of the outcomes is  statistically independent..

c2:Chi squared test summary:

method: pearson

degrees of freedom = 84

statistic = 271.79999999999995

pValue = 0.0

Very strong presumption against null hypothesis: the occurrence of the outcomes is  statistically independent..

c2:Chi squared test summary:

method: pearson

degrees of freedom = 42statistic = 271.75

pValue = 0.0

Very strong presumption against null hypothesis: the occurrence of the outcomes is  statistically independent..

---------------------------------------------------------------------------------------------------------------------------------------------------

**/

这里实际上是把特征数据中的每一列都与标签进行独立性检验。可以看出，P值都非常小，说明可以拒绝“某列与标签列无关”的假设。也就是说，可以认为每一列的数据都与最后的标签有相关性。

五、随机数生成 random data generation

RandomRDDs 是一个工具集，用来生成含有随机数的RDD，可以按各种给定的分布模式生成数据集，Random RDDs包下现支持正态分布、泊松分布和均匀分布三种分布方式。RandomRDDs提供随机double RDDS或vector RDDS。

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaDoubleRDD;

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.api.java.JavaSparkContext;

import org.apache.spark.mllib.random.RandomRDDs;

JavaDoubleRDD u = RandomRDDs.normalJavaRDD(sc, 1000000L, 10); //生成1000000个服从正态分配N(0,1)的RDD[Double]，并且分布在 10 个分区中：

JavaDoubleRDD v = u.mapToDouble(x->1.0+2.0*x); //把生成的随机数转化成N(1,4) 正态分布：

六、核密度估计 Kernel density estimation

Spark ML 提供了一个工具类 KernelDensity 用于核密度估算，核密度估算的意思是根据已知的样本估计未知的密度，属於非参数检验方法之一。核密度估计的原理是。观察某一事物的已知分布，如果某一个数在观察中出现了，可认为这个数的概率密度很大，和这个数比较近的数的概率密度也会比较大，而那些离这个数远的数的概率密度会比较小。Spark1.6.2版本支持高斯核(Gaussian kernel)。

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaRDD;

import  org.apache.spark.api.java.JavaSparkContext;

import  org.apache.spark.mllib.stat.KernelDensity;

JavaRDD<Double> d = source.map(t ->  Double.parseDouble(t.split(",")[0])); //用样本数据构建核函数，这里用假设检验中得到的iris的第一个属性的数据作为样本数据进行估计：

KernelDensity kd = new  KernelDensity().setBandwidth(3.0).setSample(d); //其中setBandwidth表示高斯核的宽度，为一个平滑参数，可以看做是高斯核的标准差。

double[] densities = kd.estimate(new double[]  {-1.0, 2.0, 5.0, 5.8});

System.out.println(Arrays.toString(densities));

/***

控制台输出结果：

------------------------------------------------------------------------------------------------------------------------------------

[0.011372003554433527, 0.05992591135719891, 0.12365409462424529,  0.12816280708978112]

-------------------------------------------------------------------------------------------------------------------------------------

**/