Sparkを使った中央値・分位点の求め方

質問

の中央値を求めるにはどうしたらよいですか？ RDD の中央値を求めることができますか？また RDD は約 70 万の要素であるため、収集して中央値を求めるには大きすぎます。

この質問と似ています。しかし、質問の答えはScalaを使用しており、私は知りません。

Apache Sparkで正確な中央値を計算するにはどうすればよいですか？

Scalaの回答の考え方を使って、Pythonで同様の回答を書こうとしています。

私は最初に RDD . 私は方法がわからない。私は sortBy (この RDD を与えられた keyfunc ) と sortByKey (これをソートして RDD は、(キー、値)のペアで構成されていると仮定しています)メソッドです。どちらもキー・バリューを使うと思うのですが、私の RDD は整数の要素しか持っていません。

1. まず、私が考えていたのは myrdd.sortBy(lambda x: x) ?
2. 次に、rddの長さを求めます( rdd.count() ).
3. 最後に、rddの中心にある要素または2つの要素を見つけたい。この方法についても助けが必要です。

EDITです。

思いつきました。多分、私は私の RDD そして、キー=インデックスと値=要素です。そして、私は値でソートしようとすることができますか？私はそれが可能であるかどうかわからない。 sortByKey メソッドしかないので、これが可能かどうかはわかりません。

どのように解決するのですか？

継続的な作業

SPARK-30569 - percentile_approx を呼び出す DSL 関数の追加

Spark 2.0+:

を使用することができます。 approxQuantile を実装したメソッドです。 Greenwald-Khannaアルゴリズム :

Python :

df.approxQuantile("x", [0.5], 0.25)

Scala :

df.stat.approxQuantile("x", Array(0.5), 0.25)

ここで、最後のパラメータは相対誤差である。数値が小さいほど、より正確な結果が得られ、より高価な計算が可能になります。

Spark 2.2以降( SPARK-14352 )からは、複数カラムでの推定をサポートしています。

df.approxQuantile(["x", "y", "z"], [0.5], 0.25)

と

df.approxQuantile(Array("x", "y", "z"), Array(0.5), 0.25)

基礎となるメソッドは、SQL集計（グローバルとグロッピングの両方）において approx_percentile 関数を使うことができます。

> SELECT approx_percentile(10.0, array(0.5, 0.4, 0.1), 100);
 [10.0,10.0,10.0]
> SELECT approx_percentile(10.0, 0.5, 100);
 10.0

スパーク 2.0

Python

コメントで述べたように、それはすべての大騒ぎをする価値がない可能性が高いです。あなたのケースのようにデータが比較的小さい場合は、単にローカルで収集し、中央値を計算します。

import numpy as np

np.random.seed(323)
rdd = sc.parallelize(np.random.randint(1000000, size=700000))

%time np.median(rdd.collect())
np.array(rdd.collect()).nbytes

私の数年前のパソコンと5.5MB程度のメモリで0.01秒程度で表示されます。

もしデータがもっと大きければ、ソートが制限要因になるので、正確な値を得る代わりに、サンプリング、収集、ローカルでの計算を行う方が良いでしょう。しかし、もし本当にSparkを使いたいのであれば、このようなものがトリックになるでしょう（もし私が何かを台無しにしていなければ）。

from numpy import floor
import time

def quantile(rdd, p, sample=None, seed=None):
    """Compute a quantile of order p ∈ [0, 1]
    :rdd a numeric rdd
    :p quantile(between 0 and 1)
    :sample fraction of and rdd to use. If not provided we use a whole dataset
    :seed random number generator seed to be used with sample
    """
    assert 0 <= p <= 1
    assert sample is None or 0 < sample <= 1

    seed = seed if seed is not None else time.time()
    rdd = rdd if sample is None else rdd.sample(False, sample, seed)

    rddSortedWithIndex = (rdd.
        sortBy(lambda x: x).
        zipWithIndex().
        map(lambda (x, i): (i, x)).
        cache())

    n = rddSortedWithIndex.count()
    h = (n - 1) * p

    rddX, rddXPlusOne = (
        rddSortedWithIndex.lookup(x)[0]
        for x in int(floor(h)) + np.array([0L, 1L]))

    return rddX + (h - floor(h)) * (rddXPlusOne - rddX)

そしていくつかのテスト。

np.median(rdd.collect()), quantile(rdd, 0.5)
## (500184.5, 500184.5)
np.percentile(rdd.collect(), 25), quantile(rdd, 0.25)
## (250506.75, 250506.75)
np.percentile(rdd.collect(), 75), quantile(rdd, 0.75)
(750069.25, 750069.25)

最後に中央値を定義します。

from functools import partial
median = partial(quantile, p=0.5)

ここまでは良いのですが、ネットワーク通信を行わないローカルモードでは4.66秒かかっています。おそらくこれを改善する方法があるはずですが、なぜ悩むのでしょうか？

言語に依存しない ( ハイブUDAF ):

もしあなたが HiveContext を使えば、HiveのUDAFも使えます。積分値で。

rdd.map(lambda x: (float(x), )).toDF(["x"]).registerTempTable("df")

sqlContext.sql("SELECT percentile_approx(x, 0.5) FROM df")

連続した値で

sqlContext.sql("SELECT percentile(x, 0.5) FROM df")

で percentile_approx では、使用するレコードの数を決定する追加の引数を渡すことができます。