このソートプロセスをpysparkで実行する方法は？

Question

map、mapValues、sortを試しましたが、何も動作しません。 質問は次のように記述されます： "類似性（値の2番目のもの）、同じ場合は、最小のID（値の最初のもの）を持つユーザーを選択してください。 そして、キーと値のペアのリストは、次のとおりです。

[ 
    (18, [(2, 0.5)]), 
    (30, [(19, 0.5), (6, 0.25)]), 
    (6, [(30, 0.25), (20, 0.2), (19, 0.2)]), 
    (19, [(30, 0.5), (8, 0.2), (6, 0.2)]), 
    (2, [(18, 0.5)]), 
    (26, [(9, 0.2)]), 
    (9, [(26, 0.2)]) 
] 

私が取得したい：

[ 
    (18, [(2, 0.5)]), 
    (30, [(19, 0.5), (6, 0.25)]), 
    (6, [(30, 0.25), (19, 0.2)]), 
    (19, [(30, 0.5), (6, 0.2)]), 
    (2, [(18, 0.5)]), 
    (26, [(9, 0.2)]), 
    (9, [(26, 0.2)]) 
] 

はあなたにたくさんありがとうございました！

Answer 1

かなりまっすぐです。必要な変換を把握するだけです。

data = [(18, [(2, 0.5)]), 
(30, [(19, 0.5), (6, 0.25)]), 
(6, [(30, 0.25), (20, 0.2), (19, 0.2)]), 
(19, [(30, 0.5), (8, 0.2), (6, 0.2)]), 
(2, [(18, 0.5)]), 
(26, [(9, 0.2)]), 
(9, [(26, 0.2)])] 

rdd1 = sc.parallelize(data) 

rdd2 = rdd1.flatMapValues(lambda x:x) 

rdd3 = rdd2.map(lambda x: ((x[0], x[1][1]),x[1][0])) 

rdd4 = rdd3.reduceByKey(min) 

rdd5 = rdd4.map(lambda x: (x[0][0], (x[1], x[0][1]))) 

rdd6 = rdd5.reduceByKey(lambda x,y: [x,y]) 
rdd6.collect() 
[(9, (26, 0.2)), 
(26, (9, 0.2)), 
(18, (2, 0.5)), 
(30, [(6, 0.25), (19, 0.5)]), 
(2, (18, 0.5)), 
(6, [(30, 0.25), (19, 0.2)]), 
(19, [(30, 0.5), (6, 0.2)])]