サイズでのPythonを経てスパークして

Question

私の100メートルを、私bigdataを準備し、量子化データ：

(1424411938', [3885, 7898]) 
(3333333333', [3885, 7898]) 

望ましい結果：

(3885, [3333333333, 1424411938]) 
(7898, [3333333333, 1424411938]) 

だから私は何をしたい、そのIグループ3885にデータを変換することです（例えば）それを持っているdata[0]すべてで）。

def prepare(data): 
    result = [] 
    for point_id, cluster in data: 
     for index, c in enumerate(cluster): 
      found = 0 
      for res in result: 
       if c == res[0]: 
        found = 1 
      if(found == 0): 
       result.append((c, [])) 
      for res in result: 
       if c == res[0]: 
        res[1].append(point_id) 
    return result 

けどmapPartitions()は「prepare()でdata RDDエドとき、このように必要以上に大きな結果を返す、私は現在のパーティションにやりたいように見える：ここで私はpythonに何をしたかです。スタートでの第一レコードは第一パーティションと2で2番目にあった場合

たとえば、私は結果として得るでしょう：

(3885, [3333333333]) 
(7898, [3333333333]) 
(3885, [1424411938]) 
(7898, [1424411938]) 

希望得るために、私prepare()を変更する方法効果？または、prepare()が生成する結果を処理する方法は、私は希望の結果を得ることができますか？

---

コードから気づいたことがあるように、私はスピードについてはまったく気にしません。ここで

は、データを作成するための方法である：

data = [] 
from random import randint 
for i in xrange(0, 10): 
    data.append((randint(0, 100000000), (randint(0, 16000), randint(0, 16000)))) 
data = sc.parallelize(data) 

Answer 1

あなたはこれを達成するための基本的なpyspark変換の束を使用することができます。

>>> rdd = sc.parallelize([(1424411938, [3885, 7898]),(3333333333, [3885, 7898])]) 
>>> r = rdd.flatMap(lambda x: ((a,x[0]) for a in x[1])) 

我々はx[1]で、鍵を持っているすべての項目の値のペアをflatMapを使用して、我々は(a, x[0])にデータラインの形式を変更し、ここでaは、すべての項目はx[1]です。 flatMapをよく理解するには、ドキュメントを参照してください。

>>> r2 = r.groupByKey().map(lambda x: (x[0],tuple(x[1]))) 

すべてのキーと値のペアをキーでグループ化し、タプル関数を使用して繰り返し可能なものをタプルに変換しました。

>>> r2.collect() 
[(3885, (1424411938, 3333333333)), (7898, (1424411938, 3333333333))] 

あなたが使用することができます言ったように、[150]最初の150の要素を持つことが、私はこれが正しい使い方だろうと思います。

r2 = r.groupByKey().map(lambda x: (x[0],tuple(x[1])[:150]))

私はできるだけ説明することを試みました。私はこれが役立つことを願っています