PySparkで爆発の反対を行う方法は？

Question

のは、私は、彼らが書いた言葉のためにユーザーのための列と別の列とDataFrameを持っているとしましょう：

Row(user='Bob', word='hello') 
Row(user='Bob', word='world') 
Row(user='Mary', word='Have') 
Row(user='Mary', word='a') 
Row(user='Mary', word='nice') 
Row(user='Mary', word='day') 

私はベクターにword列を集約したいと思います：

Row(user='Bob', words=['hello','world']) 
Row(user='Mary', words=['Have','a','nice','day']) 

それを後の集約ステップを期待しているので、私はSparksのグループ化関数を使用することはできないようです。私は、これらのデータをWord2Vecにフィードして、他のSparkアグリゲーションを使用しないようにしています。

Answer 1

ここにはrddを使用する解決策があります。

from pyspark.sql import Row 
rdd = spark.sparkContext.parallelize([Row(user='Bob', word='hello'), 
             Row(user='Bob', word='world'), 
             Row(user='Mary', word='Have'), 
             Row(user='Mary', word='a'), 
             Row(user='Mary', word='nice'), 
             Row(user='Mary', word='day')]) 
group_user = rdd.groupBy(lambda x: x.user) 
group_agg = group_user.map(lambda x: Row(**{'user': x[0], 'word': [t.word for t in x[1]]})) 

group_agg.collect()から出力：

[Row(user='Bob', word=['hello', 'world']), 
Row(user='Mary', word=['Have', 'a', 'nice', 'day'])] 

Answer 2

RDDソリューションを与えるため@titipatに感謝します。

from pyspark.sql import Row 
from pyspark.sql.functions import collect_set 
rdd = spark.sparkContext.parallelize([Row(user='Bob', word='hello'), 
             Row(user='Bob', word='world'), 
             Row(user='Mary', word='Have'), 
             Row(user='Mary', word='a'), 
             Row(user='Mary', word='nice'), 
             Row(user='Mary', word='day')]) 
df = spark.createDataFrame(rdd) 
group_user = df.groupBy('user').agg(collect_set('word').alias('words')) 
print(group_user.collect()) 

>[Row(user='Mary', words=['Have', 'nice', 'day', 'a']), Row(user='Bob', words=['world', 'hello'])] 

Answer 3

from pyspark.sql import functions as F 

df.groupby("user").agg(F.collect_list("word))