Pyspark: упорядочить / отсортировать к тому моменту - PullRequest
Новая
       6

Pyspark: упорядочить / отсортировать к тому моменту

1 голос
/ 24 апреля 2020

У меня есть такой фрейм данных: 

   usr     sec    scrpt
0  1        5     This
1  2        10      is
2  3        12       a
3  1        7    string
4  2        4      oreo

Я пытаюсь упорядочить / отсортировать по пользователю, se c, а затем сгруппировать его по пользователю и объединить там строку. так что эта таблица состоит из каждого звонка пользователя, за который он говорил. итоговый фрейм данных должен выглядеть так: 

user   concated
1      this string
2      oreo is
3      a

Я пробовал ниже в python и отлично работает 

df.sort_values(by=['usr','sec'],ascending=[True, True]).groupby(['usr')['scrpt'].apply(lambda x: ','.join(x)).reset_index()

Может кто-нибудь дать мне подобное в pyspark?

Ответы [ 2 ]

1 голос
/ 24 апреля 2020

Вы можете использовать Window функциональность для достижения sh того, что вы хотите в PySpark. 

    import pyspark.sql.functions as sf

    # Construct a window to construct sentences
    sentence_window = Window.partitionBy('usr').orderBy(sf.col('sec').asc())

    # Construct a window to get the last sentence. The others will be sentence fragments spoken by the user.
    rank_window = Window.partitionBy('usr').orderBy(sf.col('sec').desc())

    user_sentences = spark_data_df.select('usr',
                                      sf.collect_list(sf.col('scrpt')).over(sentence_window).alias('sentence'),
                                      sf.rank().over(rank_window).alias('rank'))

     user_sentences = user_sentences.filter("rank = 1").drop('rank')
     user_sentences = user_sentences.withColumn('sentence', sf.concat_ws(' ', sf.col('sentence')))

     user_sentences.show(10, False)
1 голос
/ 24 апреля 2020

С Spark-2.4+ используйте функции array_join, sort_array, transform для этого случая. 

#sample dataframe

df=spark.createDataFrame([(1,5,"This"),(2,10,"is"),(3,12,"a"),(1,7,"string"),(2,4,"oreo")],["usr","sec","scrpt"])

df.show()
#+---+---+------+
#|usr|sec| scrpt|
#+---+---+------+
#|  1|  5|  This|
#|  2| 10|    is|
#|  3| 12|     a|
#|  1|  7|string|
#|  2|  4|  oreo|
#+---+---+------+

df.groupBy("usr").agg(array_join(expr("""transform(sort_array(collect_list(struct(sec,scrpt)),True), x -> x.scrpt)""")," ").alias("concated")).orderBy("usr").show(10,False)

df.groupBy("usr").agg(concat_ws(" ",expr("""transform(sort_array(collect_list(struct(sec,scrpt)),True), x -> x.scrpt)""")).alias("concated")).orderBy("usr").show(10,False)
#+---+-----------+
#|usr|concated   |
#+---+-----------+
#|1  |This string|
#|2  |oreo is    |
#|3  |a          |
#+---+-----------+

#lower case
df.groupBy("usr").agg(lower(array_join(expr("""transform(sort_array(collect_list(struct(sec,scrpt)),True), x -> x.scrpt)""")," ")).alias("concated")).orderBy("usr").show(10,False)
#+---+-----------+
#|usr|concated   |
#+---+-----------+
#|1  |this string|
#|2  |oreo is    |
#|3  |a          |
#+---+-----------+
...