In PySpark dataframe - Concatenate String values for multiple rows and group by Project id

user_16243239 0
Viewed 467

I am new to AWS Glue, Python and PySparK. In my dataset, I am creating a Glue job with PySpark dataframe that will perform a concat distinct then group by while forming unique rows. I have 2 columns, 'project_id' and 'item'. I need to group by 'Project_ID, then show 'Item' values concatenated into the unique 'Project_ID' rows. Data sample:

Project_ID  Item
10291318    204281
10291318    204284
10217060    106934
10217060    106934
10217060    106935
             46439
             47325
             51862
10225836    118100
10225836    289841
10262244    163491
10262244    271251
10262244    271252

Expected result:

Project_ID  Item
10291318    204281, 204284
10217060    106934, 106935
             46439
             47325
             51862
10225836    118100, 289841
10262244    163491, 271251, 271252

Here is my dataframe, PySpark code:

def GroupRows(glueContext, dfc) -> DynamicFrameCollection:
    df = dfc.select(list(dfc.keys())[0]).toDF()
    
    from awsglue.dynamicframe import DynamicFrame
    from pyspark.sql import functions as f
    import datetime

    df=df.groupby("Project ID")\
        .agg(f.concat_ws(',',f.collect_set("Item")).alias("Item"))

    return DynamicFrameCollection({"ct0":dfc}, glueContext)

I have looked up numerous examples but none seem to provide the expected result. Any help would be much appreciated.

pyspark
2 Answers

There are many ways to achieve this.

  • You could split your dataframe by filter, group by into the not null Project_ID dataframe and union them.
  • Self join with filters.
  • Window and Distinct

The example for the last one is:

from pyspark.sql import functions as f
from pyspark.sql import Window
import datetime

w = Window.partitionBy("Project_ID")

df = spark.read.option("inferSchema","true").option("header","true").csv("test.csv")
df = df.withColumn("Item", f.when(f.expr("Project_ID is not null"), f.concat_ws(',' , f.collect_set("Item").over(w))).otherwise(f.col("Item"))).distinct()

+----------+--------------------+
|Project_ID|Item                |
+----------+--------------------+
|null      |46439               |
|null      |47325               |
|null      |51862               |
|10217060  |106934,106935       |
|10291318  |204284,204281       |
|10262244  |163491,271251,271252|
|10225836  |289841,118100       |
+----------+--------------------+
user_118415710

You can use collect_set or collect_list. Collect list will give you unique items

import pyspark.sql.functions as F
(df.groupby("Project_ID").agg(F.collect_set("Item"),F.collect_list("Item")).show())

+----------+-----------------+------------------+
|Project_ID|collect_set(Item)|collect_list(Item)|
+----------+-----------------+------------------+
|  10291318|         [23, 11]|          [11, 23]|
|  10217060|     [56, 21, 22]|      [22, 56, 21]|
+----------+-----------------+------------------+

Here is the sample data I used

df = spark.createDataFrame(
    [ (10291318, "11"),   (10291318, "23"),   (10217060, "21"),  (10217060, "56"), (10217060, "22")  ],  ["Project_ID", "Item"] )
user_58547310
Related