基本統計量 - 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}} \]

記述統計

RDD[Vector]に対する列ごとの記述統計量は、Statisticsで利用可能なcolStats関数を通して提供されます。

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])]
)  # an RDD of Vectors

# Compute column summary statistics.
summary = Statistics.colStats(mat)
print(summary.mean())  # a dense vector containing the mean value for each column
print(summary.variance())  # column-wise variance
print(summary.numNonzeros())  # number of nonzeros in each column

完全なコード例は、Sparkリポジトリの"examples/src/main/python/mllib/summary_statistics_example.py"にあります。

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)
  )
)

// Compute column summary statistics.
val summary: MultivariateStatisticalSummary = Statistics.colStats(observations)
println(summary.mean)  // a dense vector containing the mean value for each column
println(summary.variance)  // column-wise variance
println(summary.numNonzeros)  // number of nonzeros in each column

完全なコード例は、Sparkリポジトリの"examples/src/main/scala/org/apache/spark/examples/mllib/SummaryStatisticsExample.scala"にあります。

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)
  )
); // an RDD of Vectors

// Compute column summary statistics.
MultivariateStatisticalSummary summary = Statistics.colStats(mat.rdd());
System.out.println(summary.mean());  // a dense vector containing the mean value for each column
System.out.println(summary.variance());  // column-wise variance
System.out.println(summary.numNonzeros());  // number of nonzeros in each column

完全なコード例は、Sparkリポジトリの"examples/src/main/java/org/apache/spark/examples/mllib/JavaSummaryStatisticsExample.java"にあります。

相関

2つのデータ系列間の相関を計算することは、統計学における一般的な操作です。spark.mllibでは、多くの系列間のペアワイズ相関を計算する柔軟性が提供されています。現在サポートされている相関方法は、ピアソンの相関とスピアマンの相関です。

Statisticsは、系列間の相関を計算するメソッドを提供します。入力の種類に応じて、2つのRDD[Double]またはRDD[Vector]の場合、出力はそれぞれDoubleまたは相関Matrixになります。

APIの詳細については、StatisticsのPythonドキュメントを参照してください。

from pyspark.mllib.stat import Statistics

seriesX = sc.parallelize([1.0, 2.0, 3.0, 3.0, 5.0])  # a series
# seriesY must have the same number of partitions and cardinality as seriesX
seriesY = sc.parallelize([11.0, 22.0, 33.0, 33.0, 555.0])

# Compute the correlation using Pearson's method. Enter "spearman" for Spearman's method.
# If a method is not specified, Pearson's method will be used by default.
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])]
)  # an RDD of Vectors

# calculate the correlation matrix using Pearson's method. Use "spearman" for Spearman's method.
# If a method is not specified, Pearson's method will be used by default.
print(Statistics.corr(data, method="pearson"))

完全なコード例は、Sparkリポジトリの"examples/src/main/python/mllib/correlations_example.py"にあります。

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))  // a series
// must have the same number of partitions and cardinality as seriesX
val seriesY: RDD[Double] = sc.parallelize(Array(11, 22, 33, 33, 555))

// compute the correlation using Pearson's method. Enter "spearman" for Spearman's method. If a
// method is not specified, Pearson's method will be used by default.
val correlation: Double = Statistics.corr(seriesX, seriesY, "pearson")
println(s"Correlation is: $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))
)  // note that each Vector is a row and not a column

// calculate the correlation matrix using Pearson's method. Use "spearman" for Spearman's method
// If a method is not specified, Pearson's method will be used by default.
val correlMatrix: Matrix = Statistics.corr(data, "pearson")
println(correlMatrix.toString)

完全なコード例は、Sparkリポジトリの"examples/src/main/scala/org/apache/spark/examples/mllib/CorrelationsExample.scala"にあります。

Statisticsは、系列間の相関を計算するメソッドを提供します。入力の種類に応じて、2つのJavaDoubleRDDまたはJavaRDD<Vector>の場合、出力はそれぞれDoubleまたは相関Matrixになります。

APIの詳細については、StatisticsのJavaドキュメントを参照してください。

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));  // a series

// must have the same number of partitions and cardinality as seriesX
JavaDoubleRDD seriesY = jsc.parallelizeDoubles(
  Arrays.asList(11.0, 22.0, 33.0, 33.0, 555.0));

// compute the correlation using Pearson's method. Enter "spearman" for Spearman's method.
// If a method is not specified, Pearson's method will be used by default.
double correlation = Statistics.corr(seriesX.srdd(), seriesY.srdd(), "pearson");
System.out.println("Correlation is: " + correlation);

// note that each Vector is a row and not a column
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)
  )
);

// calculate the correlation matrix using Pearson's method.
// Use "spearman" for Spearman's method.
// If a method is not specified, Pearson's method will be used by default.
Matrix correlMatrix = Statistics.corr(data.rdd(), "pearson");
System.out.println(correlMatrix.toString());

完全なコード例は、Sparkリポジトリの"examples/src/main/java/org/apache/spark/examples/mllib/JavaCorrelationsExample.java"にあります。

層化サンプリング

他の統計関数（spark.mllibにある）とは異なり、層化サンプリングメソッドであるsampleByKeyとsampleByKeyExactは、キーバリューペアのRDDに対して実行できます。層化サンプリングでは、キーはラベル、値は特定の属性と考えることができます。例えば、キーは男性または女性、または文書IDであり、対応する値は人口における人々の年齢のリストまたは文書内の単語のリストです。sampleByKeyメソッドは、観測値がサンプリングされるかどうかを決定するためにコイン投げを行い、そのためデータ上を1回通過するだけで済み、*期待される*サンプルサイズを提供します。sampleByKeyExactは、sampleByKeyで使用される層ごとの単純無作為サンプリングよりも大幅に多くのリソースを必要としますが、99.99％の信頼度で正確なサンプルサイズを提供します。sampleByKeyExactは現在Pythonではサポートされていません。

sampleByKey()を使用すると、$f_k$がキー$k$の目的の割合、$n_k$がキー$k$のキーバリューペアの数、$K$がキーの集合である場合、$\lceil f_k \cdot n_k \rceil \, \forall k \in K$個のアイテムを近似的にサンプリングできます。

注記：sampleByKeyExact()は現在Pythonではサポートされていません。

# an RDD of any key value pairs
data = sc.parallelize([(1, 'a'), (1, 'b'), (2, 'c'), (2, 'd'), (2, 'e'), (3, 'f')])

# specify the exact fraction desired from each key as a dictionary
fractions = {1: 0.1, 2: 0.6, 3: 0.3}

approxSample = data.sampleByKey(False, fractions)

完全なコード例は、Sparkリポジトリの"examples/src/main/python/mllib/stratified_sampling_example.py"にあります。

sampleByKeyExact()を使用すると、$f_k$がキー$k$の目的の割合、$n_k$がキー$k$のキーバリューペアの数、$K$がキーの集合である場合、$\lceil f_k \cdot n_k \rceil \, \forall k \in K$個のアイテムを正確にサンプリングできます。非復元サンプリングは、サンプルサイズを保証するためにRDD上をさらに1回通過する必要があるのに対し、復元サンプリングはさらに2回通過する必要があります。

// an RDD[(K, V)] of any key value pairs
val data = sc.parallelize(
  Seq((1, 'a'), (1, 'b'), (2, 'c'), (2, 'd'), (2, 'e'), (3, 'f')))

// specify the exact fraction desired from each key
val fractions = Map(1 -> 0.1, 2 -> 0.6, 3 -> 0.3)

// Get an approximate sample from each stratum
val approxSample = data.sampleByKey(withReplacement = false, fractions = fractions)
// Get an exact sample from each stratum
val exactSample = data.sampleByKeyExact(withReplacement = false, fractions = fractions)

完全なコード例は、Sparkリポジトリの"examples/src/main/scala/org/apache/spark/examples/mllib/StratifiedSamplingExample.scala"にあります。

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);

// specify the exact fraction desired from each key Map<K, Double>
ImmutableMap<Integer, Double> fractions = ImmutableMap.of(1, 0.1, 2, 0.6, 3, 0.3);

// Get an approximate sample from each stratum
JavaPairRDD<Integer, Character> approxSample = data.sampleByKey(false, fractions);
// Get an exact sample from each stratum
JavaPairRDD<Integer, Character> exactSample = data.sampleByKeyExact(false, fractions);

完全なコード例は、Sparkリポジトリの"examples/src/main/java/org/apache/spark/examples/mllib/JavaStratifiedSamplingExample.java"にあります。

仮説検定

仮説検定は、結果が統計的に有意であるかどうか、この結果が偶然発生したかどうかを判断するための統計学における強力なツールです。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)  # a vector composed of the frequencies of events

# compute the goodness of fit. If a second vector to test against
# is not supplied as a parameter, the test runs against a uniform distribution.
goodnessOfFitTestResult = Statistics.chiSqTest(vec)

# summary of the test including the p-value, degrees of freedom,
# test statistic, the method used, and the null hypothesis.
print("%s\n" % goodnessOfFitTestResult)

mat = Matrices.dense(3, 2, [1.0, 3.0, 5.0, 2.0, 4.0, 6.0])  # a contingency matrix

# conduct Pearson's independence test on the input contingency matrix
independenceTestResult = Statistics.chiSqTest(mat)

# summary of the test including the p-value, degrees of freedom,
# test statistic, the method used, and the null hypothesis.
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(label, feature)

# The contingency table is constructed from an RDD of LabeledPoint and used to conduct
# the independence test. Returns an array containing the ChiSquaredTestResult for every feature
# against the label.
featureTestResults = Statistics.chiSqTest(obs)

for i, result in enumerate(featureTestResults):
    print("Column %d:\n%s" % (i + 1, result))

完全なコード例は、Sparkリポジトリの"examples/src/main/python/mllib/hypothesis_testing_example.py"にあります。

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

// a vector composed of the frequencies of events
val vec: Vector = Vectors.dense(0.1, 0.15, 0.2, 0.3, 0.25)

// compute the goodness of fit. If a second vector to test against is not supplied
// as a parameter, the test runs against a uniform distribution.
val goodnessOfFitTestResult = Statistics.chiSqTest(vec)
// summary of the test including the p-value, degrees of freedom, test statistic, the method
// used, and the null hypothesis.
println(s"$goodnessOfFitTestResult\n")

// a contingency matrix. Create a dense matrix ((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))

// conduct Pearson's independence test on the input contingency matrix
val independenceTestResult = Statistics.chiSqTest(mat)
// summary of the test including the p-value, degrees of freedom
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)
      )
    )
  ) // (label, feature) pairs.

// The contingency table is constructed from the raw (label, feature) pairs and used to conduct
// the independence test. Returns an array containing the ChiSquaredTestResult for every feature
// against the label.
val featureTestResults: Array[ChiSqTestResult] = Statistics.chiSqTest(obs)
featureTestResults.zipWithIndex.foreach { case (k, v) =>
  println(s"Column ${(v + 1)} :")
  println(k)
}  // summary of the test

完全なコード例は、Sparkリポジトリの"examples/src/main/scala/org/apache/spark/examples/mllib/HypothesisTestingExample.scala"にあります。

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;

// a vector composed of the frequencies of events
Vector vec = Vectors.dense(0.1, 0.15, 0.2, 0.3, 0.25);

// compute the goodness of fit. If a second vector to test against is not supplied
// as a parameter, the test runs against a uniform distribution.
ChiSqTestResult goodnessOfFitTestResult = Statistics.chiSqTest(vec);
// summary of the test including the p-value, degrees of freedom, test statistic,
// the method used, and the null hypothesis.
System.out.println(goodnessOfFitTestResult + "\n");

// Create a contingency matrix ((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});

// conduct Pearson's independence test on the input contingency matrix
ChiSqTestResult independenceTestResult = Statistics.chiSqTest(mat);
// summary of the test including the p-value, degrees of freedom...
System.out.println(independenceTestResult + "\n");

// an RDD of labeled points
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))
  )
);

// The contingency table is constructed from the raw (label, feature) pairs and used to conduct
// the independence test. Returns an array containing the ChiSquaredTestResult for every feature
// against the label.
ChiSqTestResult[] featureTestResults = Statistics.chiSqTest(obs.rdd());
int i = 1;
for (ChiSqTestResult result : featureTestResults) {
  System.out.println("Column " + i + ":");
  System.out.println(result + "\n");  // summary of the test
  i++;
}

完全なコード例は、Sparkリポジトリの"examples/src/main/java/org/apache/spark/examples/mllib/JavaHypothesisTestingExample.java"にあります。

さらに、spark.mllibは、確率分布の等しさに対するコルモゴロフ・スミルノフ（KS）検定の片側2標本実装を提供します。理論分布の名前（現在、正規分布のみサポート）とそのパラメータ、または与えられた理論分布に従って累積分布を計算する関数を指定することで、ユーザーはサンプルがその分布から抽出されたという帰無仮説を検定できます。ユーザーが正規分布に対して検定するケース（distName="norm"）、ただし分布パラメータを指定しない場合、検定は標準正規分布で初期化され、適切なメッセージがログに出力されます。

Statistics は、片側2標本コルモゴロフ・スミルノフ検定を実行するためのメソッドを提供します。次の例は、仮説検定の実行と解釈方法を示しています。

APIの詳細については、StatisticsのPythonドキュメントを参照してください。

from pyspark.mllib.stat import Statistics

parallelData = sc.parallelize([0.1, 0.15, 0.2, 0.3, 0.25])

# run a KS test for the sample versus a standard normal distribution
testResult = Statistics.kolmogorovSmirnovTest(parallelData, "norm", 0, 1)
# summary of the test including the p-value, test statistic, and null hypothesis
# if our p-value indicates significance, we can reject the null hypothesis
# Note that the Scala functionality of calling Statistics.kolmogorovSmirnovTest with
# a lambda to calculate the CDF is not made available in the Python API
print(testResult)

完全なサンプルコードは、Sparkリポジトリの "examples/src/main/python/mllib/hypothesis_testing_kolmogorov_smirnov_test_example.py"にあります。

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))  // an RDD of sample data

// run a KS test for the sample versus a standard normal distribution
val testResult = Statistics.kolmogorovSmirnovTest(data, "norm", 0, 1)
// summary of the test including the p-value, test statistic, and null hypothesis if our p-value
// indicates significance, we can reject the null hypothesis.
println(testResult)
println()

// perform a KS test using a cumulative distribution function of our making
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)

完全なサンプルコードは、Sparkリポジトリの "examples/src/main/scala/org/apache/spark/examples/mllib/HypothesisTestingKolmogorovSmirnovTestExample.scala"にあります。

APIの詳細については、StatisticsのJavaドキュメントを参照してください。

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);
// summary of the test including the p-value, test statistic, and null hypothesis
// if our p-value indicates significance, we can reject the null hypothesis
System.out.println(testResult);

完全なサンプルコードは、Sparkリポジトリの "examples/src/main/java/org/apache/spark/examples/mllib/JavaHypothesisTestingKolmogorovSmirnovTestExample.java"にあります。

ストリーミング有意性検定

spark.mllibは、A/Bテストのようなユースケースをサポートするために、いくつかの検定のオンライン実装を提供します。これらの検定は、Spark Streaming DStream[(Boolean, Double)]で実行できます。ここで、各タプルの最初の要素はコントロールグループ（false）または処理グループ（true）を示し、2番目の要素は観測値です。

ストリーミング有意性検定は、次のパラメータをサポートします。

peacePeriod - 新規性の影響を軽減するために無視される、ストリームからの初期データポイントの数。
windowSize - 仮説検定を実行する過去のバッチの数。0に設定すると、すべて以前のバッチを使用して累積処理が実行されます。

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()

完全なサンプルコードは、Sparkリポジトリの "examples/src/main/scala/org/apache/spark/examples/mllib/StreamingTestExample.scala"にあります。

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();

完全なサンプルコードは、Sparkリポジトリの "examples/src/main/java/org/apache/spark/examples/mllib/JavaStreamingTestExample.java"にあります。

乱数の生成

ランダムデータ生成は、ランダム化アルゴリズム、プロトタイピング、パフォーマンステストに役立ちます。spark.mllibは、与えられた分布（一様分布、標準正規分布、ポアソン分布）から抽出された独立同一分布（i.i.d.）の値を持つランダムRDDを生成することをサポートします。

RandomRDDs は、ランダムなdouble RDDまたはベクトルRDDを生成するためのファクトリメソッドを提供します。次の例は、値が標準正規分布N(0, 1)に従うランダムなdouble RDDを生成し、それをN(1, 4)にマッピングします。

APIの詳細については、RandomRDDs Pythonドキュメントを参照してください。

from pyspark.mllib.random import RandomRDDs

sc = ... # SparkContext

# Generate a random double RDD that contains 1 million i.i.d. values drawn from the
# standard normal distribution `N(0, 1)`, evenly distributed in 10 partitions.
u = RandomRDDs.normalRDD(sc, 1000000L, 10)
# Apply a transform to get a random double RDD following `N(1, 4)`.
v = u.map(lambda x: 1.0 + 2.0 * x)

APIの詳細については、RandomRDDs Scalaドキュメントを参照してください。

import org.apache.spark.SparkContext
import org.apache.spark.mllib.random.RandomRDDs._

val sc: SparkContext = ...

// Generate a random double RDD that contains 1 million i.i.d. values drawn from the
// standard normal distribution `N(0, 1)`, evenly distributed in 10 partitions.
val u = normalRDD(sc, 1000000L, 10)
// Apply a transform to get a random double RDD following `N(1, 4)`.
val v = u.map(x => 1.0 + 2.0 * x)

APIの詳細については、RandomRDDs Javaドキュメントを参照してください。

import org.apache.spark.SparkContext;
import org.apache.spark.api.JavaDoubleRDD;
import static org.apache.spark.mllib.random.RandomRDDs.*;

JavaSparkContext jsc = ...

// Generate a random double RDD that contains 1 million i.i.d. values drawn from the
// standard normal distribution `N(0, 1)`, evenly distributed in 10 partitions.
JavaDoubleRDD u = normalJavaRDD(jsc, 1000000L, 10);
// Apply a transform to get a random double RDD following `N(1, 4)`.
JavaDoubleRDD v = u.mapToDouble(x -> 1.0 + 2.0 * x);

カーネル密度推定

カーネル密度推定は、観測されたサンプルが抽出された特定の分布に関する仮定を必要とせずに、経験的確率分布を視覚化するために有用な手法です。これは、与えられた点集合で評価された確率変数の確率密度関数の推定値を計算します。これは、特定の点での経験的分布のPDFを、各サンプルを中心とした正規分布のPDFの平均として表現することで実現されます。

KernelDensity は、サンプルのRDDからカーネル密度推定を計算するためのメソッドを提供します。次の例は、その方法を示しています。

APIの詳細については、KernelDensity Pythonドキュメントを参照してください。

from pyspark.mllib.stat import KernelDensity

# an RDD of sample data
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])

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

# Find density estimates for the given values
densities = kd.estimate([-1.0, 2.0, 5.0])

完全なサンプルコードは、Sparkリポジトリの "examples/src/main/python/mllib/kernel_density_estimation_example.py"にあります。

KernelDensity は、サンプルのRDDからカーネル密度推定を計算するためのメソッドを提供します。次の例は、その方法を示しています。

APIの詳細については、KernelDensity Scalaドキュメントを参照してください。

import org.apache.spark.mllib.stat.KernelDensity
import org.apache.spark.rdd.RDD

// an RDD of sample data
val data: RDD[Double] = sc.parallelize(Seq(1, 1, 1, 2, 3, 4, 5, 5, 6, 7, 8, 9, 9))

// 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))

完全なサンプルコードは、Sparkリポジトリの "examples/src/main/scala/org/apache/spark/examples/mllib/KernelDensityEstimationExample.scala"にあります。

KernelDensity は、サンプルのRDDからカーネル密度推定を計算するためのメソッドを提供します。次の例は、その方法を示しています。

APIの詳細については、KernelDensity Javaドキュメントを参照してください。

import java.util.Arrays;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.mllib.stat.KernelDensity;

// an RDD of sample data
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));

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

// Find density estimates for the given values
double[] densities = kd.estimate(new double[]{-1.0, 2.0, 5.0});

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

完全なサンプルコードは、Sparkリポジトリの "examples/src/main/java/org/apache/spark/examples/mllib/JavaKernelDensityEstimationExample.java"にあります。

MLlib：メインガイド

MLlib：RDDベースのAPIガイド

基本統計量 - RDDベースのAPI

記述統計

相関

層化サンプリング

仮説検定

ストリーミング有意性検定

乱数の生成

カーネル密度推定