私はスパーク1.6.2を使用していますpySpark

Question

にグループごとに上位Nアイテムの取得、私は次のようなデータ構造を持っている：

sample = sqlContext.createDataFrame([ 
        (1,['potato','orange','orange']), 
        (1,['potato','orange','yogurt']), 
        (2,['vodka','beer','vodka']), 
        (2,['vodka','beer','juice', 'vinegar']) 

    ],['cat','terms']) 

私は猫ごとに上位N最も頻繁に用語を抽出したいと思います。私はうまくいくように見える次のソリューションを開発しましたが、これを行うより良い方法があるかどうかを見たいと思っていました。

from collections import Counter 
def get_top(it, terms=200): 
    c = Counter(it.__iter__()) 
    return [x[0][1] for x in c.most_common(terms)] 

(sample.select('cat',sf.explode('terms')).rdd.map(lambda x: (x.cat, x.col)) 
.groupBy(lambda x: x[0]) 
.map(lambda x: (x[0], get_top(x[1], 2))) 
.collect() 
) 

それは次のような出力を提供：

[(1, ['orange', 'potato']), (2, ['vodka', 'beer'])] 

私が探しています何に沿ったものであるが、私は本当に私がカウンターを使用してに頼るていたという事実が好きではありません。一人で火花を燃やしてどうすればいいですか？

おかげ

Answer 1

これは、それが動作している場合Code Reviewにこれを投稿する方が良いでしょう。

練習と同じように、私はカウンターなしでこれを行いましたが、主に同じ機能を複製するだけです。

• カウントcat
• 並べ替え（cat、term）
• グループの各出現値の用語の数（2）までカウントし、スライスに基づい

コード：

from operator import add 

(sample.select('cat', sf.explode('terms')) 
.rdd 
.map(lambda x: (x, 1)) 
.reduceByKey(add) 
.groupBy(lambda x: x[0][0]) 
.mapValues(lambda x: [r[1] for r, _ in sorted(x, key=lambda a: -a[1])[:2]]) 
.collect()) 

出力：

[(1, ['orange', 'potato']), (2, ['vodka', 'beer'])]