如何重整這段pyspark code以更有效率

ericali · **註冊時間:** 2014-12-01, 09:49 **文章:** 1

目標: 將資料欄位依照特定key產生mean, min, max, count等描述性指標到local file (目前用json format輸出)
問題: 成功讓key帶出描述指標, 但是程式很醜= =, 而且3.5G資料費時2分鐘, 希望優化code縮短時間. 根據UI的stages說明, 使用到不少shuffle write/read, 如附加檔案

資料範例: 三個欄位, key為Name+Shop, 針對Money整理出描述性指標
Name Shop Money
A Shop001 99.99
A Shop001 87.15
B Shop001 3.99
...

目前的程式, sc進來以後:

代碼:

def tupleDivide(y):
    return float(y[0])/y[1]

def smin(a, b):
    return min(a, b)

def smax(a, b):
    return max(a, b)

raw = sgRDD.map(lambda x: getVar(parserLine(x),list_C+list_N)).cache()
cnt = raw.map(lambda (x,y,z): (x+"_"+y, 1)).countByKey()
sum = raw.map(lambda (x,y,z): (x+"_"+y, z)).reduceByKey(add)
min = raw.map(lambda (x,y,z): (x+"_"+y, z)).reduceByKey(smin)
max = raw.map(lambda (x,y,z): (x+"_"+y, z)).reduceByKey(smax)
raw_cntRDD = sc.parallelize(cnt.items(),3)
raw_mean = sum.join(raw_cntRDD).map(lambda (x, y): (x, tupleDivide(y))) 

sgRDD是將data接進來
發現我使用太多重複地code, 卻不曉得該如何修改

懇請給我一些建議, 謝謝!

cfchou · **註冊時間:** 2015-02-13, 16:33 **文章:** 1

FYI

代碼:

def foo(data: RDD[(String, Double)]): RDD[(String, (Double, Double, Double))] = {
    val tmp = data map { case (s, i) =>
      (s, (i, i, i))
    }

   // (sum, max, min)
    val mm = tmp.reduceByKey({ case (v1, v2) =>
      (v1._1 + v2._1, max(v1._2, v2._2), min(v1._3, v2._3))
    })
}

Taiwan Hadoop Forum

如何重整這段pyspark code以更有效率

誰在線上