Pyspark RDDでユニークなタプルを見つけ、私はのようにpysparkにショッピングプラットフォーム上のユーザアクティビティデータのRDDを持って

Question

：

のuser_idを| product_id | イベント（商品を見る、購入した、カートに追加したなど）

同じもの（user_id、product_id）タプルには複数のイベントタイプがあります。そのような出来事をすべて同じ列に集めたいと思います。

例：

╔═════════════════════════════════════════════════╗ 
║ user_id | product_id    | Event ║ 
╠═════════════════════════════════════════════════╣ 
║ 1    1      viewed ║ 
║ 1    1      purchased ║ 
║ 2    1      added  ║ 
║ 2    2      viewed ║ 
║ 2    2      added  ║ 
╚═════════════════════════════════════════════════╝ 

私がしたい：Scalaでは

╔════════════════════════════════════════════════╗ 
║ user_id | product_id |  Event    ║ 
╠════════════════════════════════════════════════╣ 
║ 1   1   {viewed, purchased}  ║ 
║ 2   1   {added}     ║ 
║ 2   2   {viewed, added}   ║ 
╚════════════════════════════════════════════════╝ 

Answer 1

を、それは次のようになります。あなたはこれをDataframe見て試してみる必要がある場合は

val grouped : RDD[((user_id, product_id), Iterable[Event])]= rdd.map(triplet => ((triplet._1, triplet._2), triplet._3)).groupByKey() 

Answer 2

： -

import pyspark.sql.functions as F 
rdd = sc.parallelize([[1, 1, 'viewed'],[1, 1, 'purchased'],[2, 1, 'added'],[2, 2, 'viewed'],[2, 2, 'added']]) 
df = rdd.toDF(['user_id', 'product_id', 'Event']) 
df.groupby(['user_id', 'product_id']).agg(F.collect_set("Event")).show() 

の場合はrddにこのための表情を追跡するのが好き： -

rdd = sc.parallelize([[1, 1, 'viewed'],[1, 1, 'purchased'],[2, 1, 'added'],[2, 2, 'viewed'],[2, 2, 'added']]) 
rdd.groupBy(lambda x:(x[0],x[1])).map(lambda x:(x[0][0], x[0][1], map(lambda x:x[2], list(x[1])))).collect()