方法を次のようにスパーク

Question

にカンマ区切りで新しい列に入力されたデータのフォーマットを同じ列の値を連結する：

+--------------------+-------------+--------------------+ 
|   date  |  user |   product | 
+--------------------+-------------+--------------------+ 
|  2016-10-01 |  Tom |   computer | 
+--------------------+-------------+--------------------+ 
|  2016-10-01 |  Tom |   iphone | 
+--------------------+-------------+--------------------+ 
|  2016-10-01 |  Jhon |    book | 
+--------------------+-------------+--------------------+ 
|  2016-10-02 |  Tom |    pen | 
+--------------------+-------------+--------------------+ 
|  2016-10-02 |  Jhon |    milk | 
+--------------------+-------------+--------------------+ 

及び出力のフォーマットを次のように

+-----------+-----------------------+ 
|  user |  products  | 
+-----------------------------------+ 
|  Tom | computer,iphone,pen | 
+-----------------------------------+ 
|  Jhon |   book,milk | 
+-----------------------------------+ 

出力には、ユーザーが注文したすべての商品が日付順に表示されます。

スパークを使用してこれらのデータを処理したいのですが、助けてください。ありがとうございました。

Answer 1

むしろGROUPBYよりマップreduceBykey（）を組み合わせて使用​​することをお勧め..また、データを仮定すると、

#Read the data using val ordersRDD = sc.textFile("/file/path") 
val ordersRDD = sc.parallelize(List(("2016-10-01","Tom","computer"), 
    ("2016-10-01","Tom","iphone"), 
    ("2016-10-01","Jhon","book"), 
    ("2016-10-02","Tom","pen"), 
    ("2016-10-02","Jhon","milk"))) 

#group by (date, user), sort by key & reduce by user & concatenate products 
val dtusrGrpRDD = ordersRDD.map(rec => ((rec._2, rec._1), rec._3)) 
    .sortByKey().map(x=>(x._1._1, x._2)) 
    .reduceByKey((acc, v) => acc+","+v) 

#if needed, make it to DF 
scala> dtusrGrpRDD.toDF("user", "product").show() 
+----+-------------------+ 
|user|   product| 
+----+-------------------+ 
| Tom|computer,iphone,pen| 
|Jhon|   book,milk| 
+----+-------------------+ 

Answer 2

あなたは（あなたがする必要があります）HiveContextを使用している場合のpythonを使用して

例：

from pyspark.sql.functions import collect_set 

df = ... load your df ... 
new_df = df.groupBy("user").agg(collect_set("product").alias("products")) 

を使用すると、重複排除製品の結果のリストをしたくない場合は、代わりにcollect_list使用することができます。