Spark を使って中央値と四分位値を見つける方法 質問する

Spark を使って中央値と四分位値を見つける方法 質問する

RDD分散メソッド、IPython、Spark を使用して整数の中央値を見つけるにはどうすればよいですか? はRDDおよそ 700,000 要素であるため、収集して中央値を見つけるには大きすぎます。

この質問は次の質問に似ています:Apache Spark で正確な中央値を計算するにはどうすればよいですか?しかし、質問の答えは Scala ですが、私はそれを知りません。

Scala の回答の考え方を使用して、Python で同様のソリューションを記述しようとしています。

まず をソートしたいのはわかっていますRDD。方法がわかりません。 sortBy(指定された でこの RDD をソートしますkeyfunc) メソッドとsortByKey( RDD(キー、値) ペアで構成されていると想定されるこの をソートします。) メソッドがあります。どちらもキー値を使用し、 には整数要素のみがあると思いますRDD

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

編集:

アイデアがありました。インデックスを作成しRDDて、キー = インデックス、値 = 要素にすることができます。その後、値で並べ替えてみます。メソッドしかないため、これが可能かどうかはわかりませんsortByKey

ベストアンサー1

進行中の作業

スパーク-30569-percentile_approxを呼び出すDSL関数を追加する

Spark 2.0以降:

approxQuantile実装するメソッドを使用することができますグリーンワルド・カナアルゴリズム:

パイソン:

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

スカラ:

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

最後のパラメータは相対誤差です。数値が低いほど、結果はより正確になりますが、計算コストは​​高くなります。

Spark 2.2以降(スパーク-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

パイソン

コメントで述べたように、大騒ぎするほどの価値はないでしょう。あなたの場合のようにデータが比較的小さい場合は、単にローカルで収集して中央値を計算するだけです。

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.5 MB のメモリでは、約 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):

を使用する場合は、HiveContextHive 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使用するレコードの数を決定する追加の引数を渡すことができます。

おすすめ記事