ユーザアイテム格付けリストをpysparkのユーザアイテムマトリクスに変換する

Question

これは、ユーザアイテム格付けリストがパンダのデータフレームのように見える方法です。

user_item_matrix = df.pivot(index = 'user_id', columns ='item_id', values = 'rating') 

item_id aaaaaaa bbbbbbb ccccccc ddddddd 
user_id          
T   NaN  NaN  NaN  1.0 
X   5.0  NaN  NaN  NaN 
Y   NaN  2.0  NaN  NaN 
Z   NaN  NaN  5.0  NaN 

私は同じ達成するために、このアプローチをしようとしています：私はパンダで、ユーザーのアイテムマトリックスを作成し、それが唯一の本当のデータセット（約500K行）と秒のカップルを取る方法

item_id rating user_id 
0 aaaaaaa  5  X 
1 bbbbbbb  2  Y 
2 ccccccc  5  Z 
3 ddddddd  1  T 

このpysparkデータフレームの結果。

from pyspark.sql.functions import first 

df.groupby('user_id') \ 
    .pivot('item_id') \ 
    .agg(first('rating')) 

しかし、実際のデータでは完了までに時間がかかります。これを実現するよりスマートで速い方法がありますか？基本的に私はユーザ項目評価リストからユーザ項目行列を構築しようとしています。

Answer 1

これは、RDDベースの代替アプローチです。

rating_list = [['aaa',5.0,'T'],['bbb',5.0,'U'],['ccc',5.0,'V'],['ddd',5.0,'W'],['eee',5.0,'X']] 
df = sc.parallelize(rating_list).toDF(['item_id','rating','user_id']) 
df.show() 

+-------+------+-------+ 
|item_id|rating|user_id| 
+-------+------+-------+ 
| aaa| 5.0|  T| 
| bbb| 5.0|  U| 
| ccc| 5.0|  V| 
| ddd| 5.0|  W| 
| eee| 5.0|  X| 
+-------+------+-------+ 


items = df.select('item_id').rdd.map(lambda data:data.item_id).collect() 
item_len = len(items) 

def transformRating(item_id,rating,user_id): 
    rating_list = [rating if ele == item_id else None for ele in items] 
    return ([user_id]+rating_list) 

df1 = (df.rdd.map(lambda data:(data.item_id,data.rating,data.user_id)) 
      .map(lambda (item,rat,uid):transformRating(item,rat,uid)) 
      .toDF(['uid']+items)) 

df1.show() 

+---+----+----+----+----+----+ 
|uid| aaa| bbb| ccc| ddd| eee| 
+---+----+----+----+----+----+ 
| T| 5.0|null|null|null|null| 
| U|null| 5.0|null|null|null| 
| V|null|null| 5.0|null|null| 
| W|null|null|null| 5.0|null| 
| X|null|null|null|null| 5.0| 
+---+----+----+----+----+----+ 

ここで、1人のユーザーが複数のアイテムを評価する可能性があるとします。その場合、user_idに基づいてRDDを減らし、評価を結合する必要があります。 .toDFの前にreduceByKey文が1つだけ追加され、そのようなdfを取得する必要があります。