I have a dataframe

I was able to combine them like this using collect_Set method

df  = df.groupby("id").agg(f.concat_ws(", ", f.collect_list(df.col1)).alias('col1'))

But I want my final output to drop consecutive duplicate items , something like this

CodePudding user response：

w=Window.partitionBy('id')

df= (df.withColumn('index',  monotonically_increasing_id())#Create an index to orderBy
        .withColumn('index',lag('col1').over(w.orderBy('index'))).na.fill('')#Create a column to use in filter
       .where(col('col1')!=col('index')).drop('index')#filter
       .groupBy('id').agg(array_join(collect_list('col1'),',').alias('col1'))#groupby, collect_list and the array_join the outcome
     .show())

 --- -------------- 
| id|          col1|
 --- -------------- 
|  1|aa,bb,cr,qe,cr|
|  2|            yt|
|  3|            uy|
|  4|            po|
 --- --------------