MLlib：主要指南

MLlib：基于RDD的API指南

基本统计 - 基于RDD的API

\[ \newcommand{\R}{\mathbb{R}} \newcommand{\E}{\mathbb{E}} \newcommand{\x}{\mathbf{x}} \newcommand{\y}{\mathbf{y}} \newcommand{\wv}{\mathbf{w}} \newcommand{\av}{\mathbf{\alpha}} \newcommand{\bv}{\mathbf{b}} \newcommand{\N}{\mathbb{N}} \newcommand{\id}{\mathbf{I}} \newcommand{\ind}{\mathbf{1}} \newcommand{\0}{\mathbf{0}} \newcommand{\unit}{\mathbf{e}} \newcommand{\one}{\mathbf{1}} \newcommand{\zero}{\mathbf{0}} \]

摘要统计

我们通过函数 colStats 提供 RDD[Vector] 的列摘要统计，该函数可在 Statistics 中找到。

colStats() 返回一个实例 MultivariateStatisticalSummary , 其中包含按列的最大值、最小值、均值、方差和非零数的数量，以及 总计数。

有关API的更多细节，请参阅 MultivariateStatisticalSummary Python文档 。 

import numpy as np
from pyspark.mllib.stat import Statistics
mat = sc.parallelize(
[np.array([1.0, 10.0, 100.0]), np.array([2.0, 20.0, 200.0]), np.array([3.0, 30.0, 300.0])]
) # 一个包含向量的RDD

# 计算列的摘要统计信息。
summary = Statistics.colStats(mat)
print(summary.mean()) # 一个包含每列均值的密集向量
print(summary.variance()) # 按列的方差
print(summary.numNonzeros()) # 每列中的非零数量
Find full example code at "examples/src/main/python/mllib/summary_statistics_example.py" in the Spark repo.

colStats() 返回一个实例的 MultivariateStatisticalSummary ， 其中包含列的最大值、最小值、均值、方差以及非零数的数量，以及 总计数。

有关API的详细信息，请参阅 MultivariateStatisticalSummary Scala文档 。 

import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.stat.{MultivariateStatisticalSummary, Statistics}
val observations = sc.parallelize(
Seq(
Vectors.dense(1.0, 10.0, 100.0),
Vectors.dense(2.0, 20.0, 200.0),
Vectors.dense(3.0, 30.0, 300.0)
)
)
// 计算列摘要统计信息。
val summary: MultivariateStatisticalSummary = Statistics.colStats(observations)
println(summary.mean) // 一个包含每列均值的稠密向量
println(summary.variance) // 按列方差
println(summary.numNonzeros) // 每列中非零元素的数量
Find full example code at "examples/src/main/scala/org/apache/spark/examples/mllib/SummaryStatisticsExample.scala" in the Spark repo.

colStats() 返回一个 MultivariateStatisticalSummary 的实例， 其中包含每列的最大值、最小值、均值、方差和非零值的数量，以及总计数。

有关API的详细信息，请参阅 MultivariateStatisticalSummary Java文档 。 

import java.util.Arrays;
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;
JavaRDD<Vector> mat = jsc.parallelize(
Arrays.asList(
Vectors.dense(1.0, 10.0, 100.0),
Vectors.dense(2.0, 20.0, 200.0),
Vectors.dense(3.0, 30.0, 300.0)
)
); // 一个包含向量的RDD
// 计算列的摘要统计信息。
MultivariateStatisticalSummary summary = Statistics.colStats(mat.rdd());
System.out.println(summary.mean()); // 一个包含每列均值的稠密向量
System.out.println(summary.variance()); // 列的方差
System.out.println(summary.numNonzeros()); // 每列中的非零数目
Find full example code at "examples/src/main/java/org/apache/spark/examples/mllib/JavaSummaryStatisticsExample.java" in the Spark repo.

相关性

计算两组数据之间的相关性是统计学中的一种常见操作。在 spark.mllib 中，我们提供了在多组数据中计算逐对相关性的灵活性。目前支持的相关性方法有皮尔逊相关和斯皮尔曼相关。

Statistics 提供方法来 计算系列之间的相关性。根据输入的类型，输出将是一个 Double 或相关的 Matrix ，两个 RDD[Double] 或者一个 RDD[Vector] 。

有关API的更多详细信息，请参考 Statistics Python文档 。 

from pyspark.mllib.stat import Statistics
seriesX = sc.parallelize([1.0, 2.0, 3.0, 3.0, 5.0]) # 一个序列
# seriesY 必须与 seriesX 有相同数量的分区和基数
seriesY = sc.parallelize([11.0, 22.0, 33.0, 33.0, 555.0])
# 使用 Pearson 方法计算相关性。输入 "spearman" 以使用 Spearman 方法。
# 如果未指定方法，将默认使用 Pearson 方法。
print("Correlation is: " + str(Statistics.corr(seriesX, seriesY, method="pearson")))
data = sc.parallelize(
[np.array([1.0, 10.0, 100.0]), np.array([2.0, 20.0, 200.0]), np.array([5.0, 33.0, 366.0])]
) # 一个 Vectors 的 RDD

# 使用 Pearson 方法计算相关矩阵。使用 "spearman" 以使用 Spearman 方法。
# 如果未指定方法，将默认使用 Pearson 方法。
print(Statistics.corr(data, method="pearson"))
Find full example code at "examples/src/main/python/mllib/correlations_example.py" in the Spark repo.

Statistics 提供方法来 计算序列之间的相关性。根据输入的类型，两个 RDD[Double] 或 一个 RDD[Vector] ，输出将分别是一个 Double 或相关性 Matrix 。

有关API的详细信息，请参阅 Statistics Scala文档 。 

import org.apache.spark.mllib.linalg._
import org.apache.spark.mllib.stat.Statistics
import org.apache.spark.rdd.RDD
val seriesX: RDD[Double] = sc.parallelize(Array(1, 2, 3, 3, 5)) // 一个序列
// 必须与 seriesX 具有相同数量的分区和基数
val seriesY: RDD[Double] = sc.parallelize(Array(11, 22, 33, 33, 555))
// 使用皮尔逊方法计算相关性。输入 "spearman" 以使用斯皮尔曼方法。如果未指定方法，默认使用皮尔逊方法。
val correlation: Double = Statistics.corr(seriesX, seriesY, "pearson")
println(s"相关性为: $correlation")
val data: RDD[Vector] = sc.parallelize(
Seq(
Vectors.dense(1.0, 10.0, 100.0),
Vectors.dense(2.0, 20.0, 200.0),
Vectors.dense(5.0, 33.0, 366.0))
) // 注意，每个向量是一行而不是一列
// 使用皮尔逊方法计算相关性矩阵。使用 "spearman" 以使用斯皮尔曼方法
// 如果未指定方法，默认使用皮尔逊方法。
val correlMatrix: Matrix = Statistics.corr(data, "pearson")
println(correlMatrix.toString)
Find full example code at "examples/src/main/scala/org/apache/spark/examples/mllib/CorrelationsExample.scala" in the Spark repo.

Statistics 提供了计算系列之间相关性的方法。根据输入的类型，两个 JavaDoubleRDD 或 一个 JavaRDD ，输出将分别是 Double 或相关 Matrix 。

请参阅 Statistics Java 文档 获取有关 API 的详细信息。 

import java.util.Arrays;
import org.apache.spark.api.java.JavaDoubleRDD;
import org.apache.spark.api.java.JavaRDD;
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;
JavaDoubleRDD seriesX = jsc.parallelizeDoubles(
Arrays.asList(1.0, 2.0, 3.0, 3.0, 5.0)); // 一个序列
// 必须与 seriesX 具有相同的分区数量和基数
JavaDoubleRDD seriesY = jsc.parallelizeDoubles(
Arrays.asList(11.0, 22.0, 33.0, 33.0, 555.0));
// 使用皮尔逊方法计算相关性。如果输入 "spearman"，则使用斯皮尔曼方法。
// 如果未指定方法，将默认为使用皮尔逊方法。
double correlation = Statistics.corr(seriesX.srdd(), seriesY.srdd(), "pearson");
System.out.println("Correlation is: " + correlation);
// 注意，每个 Vector 是一行，而不是一列
JavaRDD<Vector> data = jsc.parallelize(
Arrays.asList(
Vectors.dense(1.0, 10.0, 100.0),
Vectors.dense(2.0, 20.0, 200.0),
Vectors.dense(5.0, 33.0, 366.0)
)
);
// 使用皮尔逊的方法计算相关矩阵。
// 使用 "spearman" 获取斯皮尔曼的方法。
// 如果未指定方法，将默认为使用皮尔逊方法。
Matrix correlMatrix = Statistics.corr(data.rdd(), "pearson");
System.out.println(correlMatrix.toString());
Find full example code at "examples/src/main/java/org/apache/spark/examples/mllib/JavaCorrelationsExample.java" in the Spark repo.

分层抽样

与其他统计函数不同，这些函数位于 spark.mllib ，分层抽样方法 sampleByKey 和 sampleByKeyExact 可以在键值对的 RDD 上进行。对于分层抽样，键可以被视为标签，而值可以被视为特定属性。例如，键可以是男性或女性，或文档 ID，相应的值可以是人口中人的年龄列表或文档中的单词列表。 sampleByKey 方法将抛硬币决定一个观察值是否会被抽样，因此需要对数据进行一次遍历，并提供一个 预期 抽样大小。 sampleByKeyExact 需要比 sampleByKey 中使用的每层简单随机抽样更多的资源，但将提供 99.99% 的置信度下的精确抽样大小。目前 sampleByKeyExact 在 python 中不受支持。

sampleByKey() 允许用户大约抽样 $\lceil f_k \cdot n_k \rceil \, \forall k \in K$ 项，其中 $f_k$ 是键 $k$ 的期望比例， $n_k$ 是键 $k$ 的键值对数量，而 $K$ 是键的集合。

注意： sampleByKeyExact() 在Python中当前不支持。 

# 一个包含任意键值对的RDD
data = sc.parallelize([(1, 'a'), (1, 'b'), (2, 'c'), (2, 'd'), (2, 'e'), (3, 'f')])
# 指定每个键所需的确切比例作为字典
fractions = {1: 0.1, 2: 0.6, 3: 0.3}
approxSample = data.sampleByKey(False, fractions)
Find full example code at "examples/src/main/python/mllib/stratified_sampling_example.py" in the Spark repo.

sampleByKeyExact() 允许用户准确地抽样 $\lceil f_k \cdot n_k \rceil \, \forall k \in K$ 项，其中 $f_k$ 是键 $k$ 的期望比例，$n_k$ 是键 $k$ 的键值对数量，$K$ 是键的集合。 不替换抽样需要额外对 RDD 进行一次遍历以保证样本大小，而替换抽样则需要额外进行两次遍历。 

// 一个 RDD[(K, V)] 的任意键值对
val data = sc.parallelize(
Seq((1, 'a'), (1, 'b'), (2, ), (2, ), (2, ), (3, )))
// 指定每个键所需的确切比例
val fractions = Map(1 -> 0.1, 2 -> 0.6, 3 -> 0.3)
// 从每个层抽取一个近似样本
val approxSample = data.sampleByKey(withReplacement = false, fractions = fractions)
// 从每个层获取一个确切的样本
val exactSample = data.sampleByKeyExact(withReplacement = false, fractions = fractions)
Find full example code at "examples/src/main/scala/org/apache/spark/examples/mllib/StratifiedSamplingExample.scala" in the Spark repo.

sampleByKeyExact() 允许用户 准确地采样 $\lceil f_k \cdot n_k \rceil \, \forall k \in K$ 项，其中 $f_k$ 是键 $k$ 的期望 比例，$n_k$ 是键 $k$ 的键值对数量，$K$ 是键的集合。无替换采样需要对 RDD 进行一次额外的遍历以保证样本 大小，而有替换采样则需要进行两次额外的遍历。 

import java.util.*;
import scala.Tuple2;
import org.apache.spark.api.java.JavaPairRDD;
List<Tuple2<Integer, Character>> list = Arrays.asList(
new Tuple2<>(1, 'a'),
new Tuple2<>(1, 'b'),
new Tuple2<>(2, 'c'),
new Tuple2<>(2, 'd'),
new Tuple2<>(2, 'e'),
new Tuple2<>(3, 'f')
);
JavaPairRDD<Integer, Character> data = jsc.parallelizePairs(list);
// 指定每个键所需的确切比例 Map
ImmutableMap<Integer, Double> fractions = ImmutableMap.of(1, 0.1, 2, 0.6, 3, 0.3);
// 从每个层获取一个近似样本
JavaPairRDD<Integer, Character> approxSample = data.sampleByKey(false, fractions);
// 从每个层获取一个确切样本
JavaPairRDD<Integer, Character> exactSample = data.sampleByKeyExact(false, fractions);
Find full example code at "examples/src/main/java/org/apache/spark/examples/mllib/JavaStratifiedSamplingExample.java" in the Spark repo.

假设检验

假设检验是统计学中的一种强大工具，用于确定结果是否在统计上显著，以及该结果是否是偶然发生的。 spark.mllib 当前支持皮尔逊卡方（ $\chi^2$）拟合优度检验和独立性检验。输入数据类型决定了是进行拟合优度检验还是独立性检验。拟合优度检验需要输入类型为 Vector ，而独立性检验则需要输入 Matrix 。

spark.mllib 还支持输入类型 RDD[LabeledPoint] 以通过卡方独立性检验启用特征选择。

Statistics 提供方法来运行皮尔逊卡方检验。以下示例演示如何运行和解释假设检验。

有关API的更多详细信息，请参考 Statistics Python文档 。 

from pyspark.mllib.linalg import Matrices, Vectors
from pyspark.mllib.regression import LabeledPoint
from pyspark.mllib.stat import Statistics
vec = Vectors.dense(0.1, 0.15, 0.2, 0.3, 0.25) # 一个由事件频率组成的向量

# 计算拟合优度。如果未提供用于测试的第二个向量，
# 则测试将在均匀分布上运行。
goodnessOfFitTestResult = Statistics.chiSqTest(vec)
# 测试的概要，包括p值、自由度、 
# 统计量、使用的方法以及原假设。
print("%s\n" % goodnessOfFitTestResult)
mat = Matrices.dense(3, 2, [1.0, 3.0, 5.0, 2.0, 4.0, 6.0]) # 一个列联矩阵

# 对输入的列联矩阵进行皮尔逊独立性测试
independenceTestResult = Statistics.chiSqTest(mat)
# 测试的概要，包括p值、自由度， 
# 统计量、使用的方法以及原假设。
print("%s\n" % independenceTestResult)
obs = sc.parallelize(
[LabeledPoint(1.0, [1.0, 0.0, 3.0]),
LabeledPoint(1.0, [1.0, 2.0, 0.0]),
LabeledPoint(1.0, [-1.0, 0.0, -0.5])]
) # LabeledPoint(标签, 特征)

# 列联表是从LabeledPoint的RDD构建的，并用于进行
# 独立性测试。返回一个包含每个特征
# 相对于标签的ChiSquaredTestResult的数组。
featureTestResults = Statistics.chiSqTest(obs)
for i, result in enumerate(featureTestResults):
print("列 %d:\n%s" % (i + 1, result))
Find full example code at "examples/src/main/python/mllib/hypothesis_testing_example.py" in the Spark repo.

Statistics 提供了运行皮尔逊卡方检验的方法。以下示例演示了如何运行和解释假设检验。 

import org.apache.spark.mllib.linalg._
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.stat.Statistics
import org.apache.spark.mllib.stat.test.ChiSqTestResult
import org.apache.spark.rdd.RDD
// 一个由事件频率组成的向量
val vec: Vector = Vectors.dense(0.1, 0.15, 0.2, 0.3, 0.25)
// 计算拟合优度。如果没有提供第二个向量进行比较作为参数，则测试将针对均匀分布运行。
val goodnessOfFitTestResult = Statistics.chiSqTest(vec)
// 测试的摘要，包括 p 值、自由度、测试统计量、所使用的方法和原假设。
println(s"$goodnessOfFitTestResult\n")
// 一个列联矩阵。创建一个稠密矩阵 ((1.0, 2.0), (3.0, 4.0), (5.0, 6.0))
val mat: Matrix = Matrices.dense(3, 2, Array(1.0, 3.0, 5.0, 2.0, 4.0, 6.0))
// 对输入列联矩阵进行皮尔逊独立性测试
val independenceTestResult = Statistics.chiSqTest(mat)
// 测试的摘要，包括 p 值、自由度
println(s"$independenceTestResult\n")
val obs: RDD[LabeledPoint] =
sc.parallelize(
Seq(
LabeledPoint(1.0, Vectors.dense(1.0, 0.0, 3.0)),
LabeledPoint(1.0, Vectors.dense(1.0, 2.0, 0.0)),
LabeledPoint(-1.0, Vectors.dense(-1.0, 0.0, -0.5)
)
)
) // (标签，特征) 对。
// 列联表是由原始（标签，特征）对构造的，并用于进行独立性测试。返回一个数组，包含每个特征的 ChiSquaredTestResult
// 与标签进行对比。
val featureTestResults: Array[ChiSqTestResult] = Statistics.chiSqTest(obs)
featureTestResults.zipWithIndex.foreach { case (k, v) =>
println(s"列 ${(v + 1)} :")
println(k)
} // 测试的摘要
Find full example code at "examples/src/main/scala/org/apache/spark/examples/mllib/HypothesisTestingExample.scala" in the Spark repo.

Statistics 提供了运行皮尔逊卡方检验的方法。以下示例演示了如何运行和解释假设检验。

有关API的详细信息，请参阅 ChiSqTestResult Java文档 。 

import java.util.Arrays;
import org.apache.spark.api.java.JavaRDD;
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.regression.LabeledPoint;
import org.apache.spark.mllib.stat.Statistics;
import org.apache.spark.mllib.stat.test.ChiSqTestResult;
// 由事件频率组成的向量
Vector vec = Vectors.dense(0.1, 0.15, 0.2, 0.3, 0.25);
// 计算拟合优度。如果没有提供用于测试的第二个向量作为参数，
// 测试将对均匀分布进行。
ChiSqTestResult goodnessOfFitTestResult = Statistics.chiSqTest(vec);
// 测试的摘要，包括p值、自由度、测试统计量、
// 使用的方法和原假设。
System.out.println(goodnessOfFitTestResult + "\n");
// 创建一个列联矩阵 ((1.0, 2.0), (3.0, 4.0), (5.0, 6.0))
Matrix mat = Matrices.dense(3, 2, new double[]{1.0, 3.0, 5.0, 2.0, 4.0, 6.0});
// 在输入的列联矩阵上进行皮尔逊独立性测试
ChiSqTestResult independenceTestResult = Statistics.chiSqTest(mat);
// 测试的摘要，包括p值、自由度...
System.out.println(independenceTestResult + "\n");
// 一个带标签点的RDD
JavaRDD<LabeledPoint> obs = jsc.parallelize(
Arrays.asList(
new LabeledPoint(1.0, Vectors.dense(1.0, 0.0, 3.0)),
new LabeledPoint(1.0, Vectors.dense(1.0, 2.0, 0.0)),
new LabeledPoint(-1.0, Vectors.dense(-1.0, 0.0, -0.5))
)
);
// 列联表是由原始（标签，特征）对构建的，用于进行
// 独立性测试。返回一个数组，包含每个特征的ChiSquaredTestResult
// 相对于标签。
ChiSqTestResult[] featureTestResults = Statistics.chiSqTest(obs.rdd());
int i = 1;
for (ChiSqTestResult result : featureTestResults) {
System.out.println("列 " + i + ":");
System.out.println(result + "\n"); // 测试的摘要
i++;
}
Find full example code at "examples/src/main/java/org/apache/spark/examples/mllib/JavaHypothesisTestingExample.java" in the Spark repo.

此外， spark.mllib 提供了一种单样本、双侧的 Kolmogorov-Smirnov (KS) 测试实现，用于概率分布的相等性。通过提供一个理论分布的名称（目前仅支持正态分布）及其参数，或者一个函数以根据给定的理论分布计算累积分布，用户可以测试其样本来自该分布的原假设。如果用户测试正态分布（ distName="norm" ），但不提供分布参数，则测试初始化为标准正态分布，并记录适当的消息。

Statistics 提供方法来运行一个1样本、双侧的Kolmogorov-Smirnov检验。以下示例演示如何运行和解释假设检验。

有关API的更多详细信息，请参考 Statistics Python文档 。 

from pyspark.mllib.stat import Statistics
parallelData = sc.parallelize([0.1, 0.15, 0.2, 0.3, 0.25])
# 针对样本与标准正态分布进行KS检验
testResult = Statistics.kolmogorovSmirnovTest(parallelData, "norm", 0, 1)
# 测试摘要，包括p值、检验统计量和原假设
# 如果我们的p值表明显著性，我们可以拒绝原假设
# 注意，Scala中调用Statistics.kolmogorovSmirnovTest并使用
# lambda计算CDF的功能在Python API中不可用
print(testResult)
Find full example code at "examples/src/main/python/mllib/hypothesis_testing_kolmogorov_smirnov_test_example.py" in the Spark repo.

Statistics 提供了运行 1-样本、2-侧的 Kolmogorov-Smirnov 测试的方法。以下示例演示了如何运行和解释假设检验。

有关API的详细信息，请参阅 Statistics Scala文档 。 

import org.apache.spark.mllib.stat.Statistics
import org.apache.spark.rdd.RDD
val data: RDD[Double] = sc.parallelize(Seq(0.1, 0.15, 0.2, 0.3, 0.25)) // 一个样本数据的 RDD
// 对样本与标准正态分布进行 KS 测试
val testResult = Statistics.kolmogorovSmirnovTest(data, "norm", 0, 1)
// 测试的概要，包括 p 值、检验统计量和原假设，如果我们的 p 值
// 表示显著性，我们可以拒绝原假设。
println(testResult)
println()
// 使用我们自己创建的累积分布函数执行 KS 测试
val myCDF = Map(0.1 -> 0.2, 0.15 -> 0.6, 0.2 -> 0.05, 0.3 -> 0.05, 0.25 -> 0.1)
val testResult2 = Statistics.kolmogorovSmirnovTest(data, myCDF)
println(testResult2)
Find full example code at "examples/src/main/scala/org/apache/spark/examples/mllib/HypothesisTestingKolmogorovSmirnovTestExample.scala" in the Spark repo.

Statistics 提供了执行 1-样本、2-侧 Kolmogorov-Smirnov 检验的方法。以下示例演示了如何运行和解释假设检验。

请参阅 Statistics Java 文档 获取有关 API 的详细信息。 

import java.util.Arrays;
import org.apache.spark.api.java.JavaDoubleRDD;
import org.apache.spark.mllib.stat.Statistics;
import org.apache.spark.mllib.stat.test.KolmogorovSmirnovTestResult;
JavaDoubleRDD data = jsc.parallelizeDoubles(Arrays.asList(0.1, 0.15, 0.2, 0.3, 0.25));
KolmogorovSmirnovTestResult testResult =
Statistics.kolmogorovSmirnovTest(data, "norm", 0.0, 1.0);
// 测试的摘要，包括 p 值、检验统计量和零假设
// 如果我们的 p 值表示显著性，我们可以拒绝零假设
System.out.println(testResult);
Find full example code at "examples/src/main/java/org/apache/spark/examples/mllib/JavaHypothesisTestingKolmogorovSmirnovTestExample.java" in the Spark repo.

流式显著性测试

spark.mllib 提供了一些测试的在线实现，以支持使用案例 如 A/B 测试。这些测试可以在一个 Spark Streaming DStream[(Boolean, Double)] 上进行，其中每个元组的第一个元素 表示对照组 ( false ) 或处理组 ( true )，第二个元素是观察值。

流式显著性测试支持以下参数：

StreamingTest 提供流媒体假设检验。 

val data = ssc.textFileStream(dataDir).map(line => line.split(",") match {
case Array(label, value) => BinarySample(label.toBoolean, value.toDouble)
})
val streamingTest = new StreamingTest()
.setPeacePeriod(0)
.setWindowSize(0)
.setTestMethod("welch")
val out = streamingTest.registerStream(data)
out.print()
Find full example code at "examples/src/main/scala/org/apache/spark/examples/mllib/StreamingTestExample.scala" in the Spark repo.

StreamingTest 提供流式假设测试。 

import org.apache.spark.mllib.stat.test.BinarySample;
import org.apache.spark.mllib.stat.test.StreamingTest;
import org.apache.spark.mllib.stat.test.StreamingTestResult;
JavaDStream<BinarySample> data = ssc.textFileStream(dataDir).map(line -> {
String[] ts = line.split(",");
boolean label = Boolean.parseBoolean(ts[0]);
double value = Double.parseDouble(ts[1]);
return new BinarySample(label, value);
});
StreamingTest streamingTest = new StreamingTest()
.setPeacePeriod(0)
.setWindowSize(0)
.setTestMethod("welch");
JavaDStream<StreamingTestResult> out = streamingTest.registerStream(data);
out.print();
Find full example code at "examples/src/main/java/org/apache/spark/examples/mllib/JavaStreamingTestExample.java" in the Spark repo.

随机数据生成

随机数据生成对随机算法、原型设计和性能测试非常有用。 spark.mllib 支持生成从给定分布中抽取的具有独立同分布（i.i.d.）值的随机RDD：均匀分布、标准正态分布或泊松分布。

RandomRDDs 提供了生成随机双精度 RDD 或向量 RDD 的工厂方法。 以下示例生成一个随机双精度 RDD，其值遵循标准正态分布 N(0, 1) ，然后将其映射到 N(1, 4) 。

有关API的更多细节，请参阅 RandomRDDs Python文档 。 

from pyspark.mllib.random import RandomRDDs
sc = ... # Spark上下文

# 生成一个包含100万个从标准正态分布`N(0, 1)`中抽取的独立同分布值的随机双精度RDD，均匀分布在10个分区中。
u = RandomRDDs.normalRDD(sc, 1000000L, 10)
# 应用变换以获得一个遵循`N(1, 4)`的随机双精度RDD。
v = u.map(lambda x: 1.0 + 2.0 * x)

RandomRDDs 提供了生成随机双精度 RDD 或向量 RDD 的工厂方法。 以下示例生成一个随机双精度 RDD，其值遵循标准正态分布 N(0, 1) ，然后将其映射到 N(1, 4) 。

有关API的详细信息，请参考 RandomRDDs Scala文档 。 

import org.apache.spark.SparkContext
import org.apache.spark.mllib.random.RandomRDDs._
val sc: SparkContext = ...
// 生成一个包含100万个独立同分布值的随机双精度RDD，值从
// 标准正态分布 `N(0, 1)` 中抽取，均匀分布在10个分区中。
val u = normalRDD(sc, 1000000L, 10)
// 应用一个变换以获取一个遵循 `N(1, 4)` 的随机双精度RDD。
val v = u.map(x => 1.0 + 2.0 * x)

RandomRDDs 提供了生成随机双精度 RDD 或向量 RDD 的工厂方法。 以下示例生成一个随机双精度 RDD，其值遵循标准正态分布 N(0, 1) ，然后将其映射到 N(1, 4) 。

有关API的详细信息，请参阅 RandomRDDs Java文档 。 

import org.apache.spark.SparkContext;
import org.apache.spark.api.JavaDoubleRDD;
import static org.apache.spark.mllib.random.RandomRDDs.*;
JavaSparkContext jsc = ...
// 生成一个包含 100 万个 i.i.d. 值的随机双精度 RDD，值来自于
// 标准正态分布 `N(0, 1)`，均匀分布在 10 个分区中。
JavaDoubleRDD u = normalJavaRDD(jsc, 1000000L, 10);
// 应用变换以获得遵循 `N(1, 4)` 的随机双精度 RDD。
JavaDoubleRDD v = u.mapToDouble(x -> 1.0 + 2.0 * x);

核密度估计

核密度估计 是一种技术， useful for visualizing empirical probability distributions without requiring assumptions about the particular distribution that the observed samples are drawn from. 它计算随机变量的概率密度函数的估计，在给定的一组点上进行评估。它通过将实证分布在某一点的PDF表示为围绕每个样本中心的正态分布的PDF的平均值来实现此估计。

KernelDensity 提供了从样本的 RDD 计算核密度估计的方法。以下示例演示了如何做到这一点。

有关API的更多详细信息，请参阅 KernelDensity Python文档 。 

from pyspark.mllib.stat import KernelDensity
# 一个样本数据的RDD
data = sc.parallelize([1.0, 1.0, 1.0, 2.0, 3.0, 4.0, 5.0, 5.0, 6.0, 7.0, 8.0, 9.0, 9.0])
# 使用样本数据和高斯核的标准差构建密度估计器
kd = KernelDensity()
kd.setSample(data)
kd.setBandwidth(3.0)
# 查找给定值的密度估计
densities = kd.estimate([-1.0, 2.0, 5.0])
Find full example code at "examples/src/main/python/mllib/kernel_density_estimation_example.py" in the Spark repo.

KernelDensity 提供从样本的 RDD 计算核密度估计的方法。以下示例演示了如何做到这一点。

有关API的详细信息，请参阅 KernelDensity Scala文档 。 

import org.apache.spark.mllib.stat.KernelDensity
import org.apache.spark.rdd.RDD
// 一个样本数据的 RDD
val data: RDD[Double] = sc.parallelize(Seq(1, 1, 1, 2, 3, 4, 5, 5, 6, 7, 8, 9, 9))
// 使用样本数据和标准差构建密度估计器
// 用于高斯核
val kd = new KernelDensity()
.setSample(data)
.setBandwidth(3.0)
// 为给定值查找密度估计
val densities = kd.estimate(Array(-1.0, 2.0, 5.0))
Find full example code at "examples/src/main/scala/org/apache/spark/examples/mllib/KernelDensityEstimationExample.scala" in the Spark repo.

KernelDensity 提供了从样本的 RDD 计算核密度估计的方法。以下示例演示了如何做到这一点。

请参考 KernelDensity Java文档 以获取API的详细信息。 

import java.util.Arrays;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.mllib.stat.KernelDensity;
// 一组样本数据的RDD
JavaRDD<Double> data = jsc.parallelize(
Arrays.asList(1.0, 1.0, 1.0, 2.0, 3.0, 4.0, 5.0, 5.0, 6.0, 7.0, 8.0, 9.0, 9.0));
// 使用样本数据构造密度估计器
// 以及高斯核的标准差
KernelDensity kd = new KernelDensity().setSample(data).setBandwidth(3.0);
// 为给定值查找密度估计
double[] densities = kd.estimate(new double[]{-1.0, 2.0, 5.0});
System.out.println(Arrays.toString(densities));
Find full example code at "examples/src/main/java/org/apache/spark/examples/mllib/JavaKernelDensityEstimationExample.java" in the Spark repo.