pysparkフィルタリングコードのパフォーマンス解析

Question

このコードは、「Summary Statistics」（https://spark.apache.org/docs/latest/mllib-statistics.html）を編集したものです。私はちょうど コレクションから最初の行を削除するには、計算の多くを実行しているように見える

mat.zipWithIndex() is O(sizeOf mat) 
filter(lambda x: x[1] > 0) is O(sizeOf mat) 
map(lambda x: x[0]) is O(sizeOf mat) 

を：具体的に私はこれを破壊pDataI = mat.zipWithIndex().filter(lambda x: x[1] > 0).map(lambda x: x[0])

を使用して文字列と浮動小数点値の組み合わせをフィルタリングしています。

全体コード：

import numpy as np 
from pyspark.mllib.stat import Statistics 

data = [] 

data.append('c1,c2,c3') 
data.append(np.array([1.0, 10.0, 100.0])) 
data.append(np.array([2.0, 20.0, 200.0])) 
data.append(np.array([3.0, 30.0, 300.0])) 

mat = sc.parallelize(
    data 
) # an RDD of Vectors 

pDataI = mat.zipWithIndex().filter(lambda x: x[1] > 0).map(lambda x: x[0]) 
summary = Statistics.colStats(pDataI) 

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 

print pDataI.stats(); 

プリント：

[ 2. 20. 200.] 

[ 1.00000000e+00 1.00000000e+02 1.00000000e+04] 

[ 3. 3. 3.] 

(count: 3, mean: [ 2. 20. 200.], stdev: [ 0.81649658 8.16496581 81.64965809], max: [ 3. 30. 300.], min: [ 1. 10. 100.]) 

これは単に1つのパスでmatコレクションを処理するために大幅に改善することができますか？

Answer 1

私のお金のために、DataFramesは常にRDDより優れたソリューションです。実際の作業環境では、データが何らかの種類のファイルやコレクションなどに格納されていると想定します（csv、寄木張り、jsonなど）。今のところcsvを想定しよう。この場合、あなただけの次の操作を行います：

df = sqlContext.read.csv('filename.csv', header=True) 

そこからあなたがdf.describe()と要約統計量を得ることができます。

Hereはあなたが読むことができるファイルタイプやコレクションのすべてを含むDataFrameReaderのドキュメントである。