Merge duplicate records into single record in a pyspark dataframe

user_4470126 0
Viewed 1713

I have a dataframe which has duplicate rows, and i would like merge them into one single record with all distinct columns.

My code sample is as follows:

df1= sqlContext.createDataFrame([("81A01","TERR NAME 01","NJ","",""),("81A01","TERR NAME 01","","NY",""),("81A01","TERR NAME 01","","","LA"),("81A02","TERR NAME 01","CA","",""),("81A02","TERR NAME 01","","","NY")], ["zip_code","territory_name","state","state1","state2"])

The resulting dataframe is as follows:

df1.show()
+--------+--------------+-----+------+------+
|zip_code|territory_name|state|state1|state2|
+--------+--------------+-----+------+------+
|   81A01|  TERR NAME 01|   NJ|      |      |
|   81A01|  TERR NAME 01|     |    NY|      |
|   81A01|  TERR NAME 01|     |      |    LA|
|   81A02|  TERR NAME 01|   CA|      |      |
|   81A02|  TERR NAME 01|     |      |    NY|
+--------+--------------+-----+------+------+

I need to merge/consolidate duplicate records based on the zip_code, and get all different state values in one row.

Expected result:

+--------+--------------+-----+------+------+
|zip_code|territory_name|state|state1|state2|
+--------+--------------+-----+------+------+
|   81A01|  TERR NAME 01|   NJ|    NY|    LA|
|   81A02|  TERR NAME 01|   CA|      |    LA|
+--------+--------------+-----+------+------+

Am new to pyspark and not sure how to use group / joins. Can someone please help with code.

pyspark apache-spark-sql python-2.7
2 Answers

if you are sure that there is only 1 state, 1 state1 and 1 state2 for each zip_code territory combination, you can use the following code. The max function uses the string, if there is a string in the grouped data, since a non-empty string has a higher value (probably ASCII wise) then the empty string ""

from pyspark.sql.types import *
from pyspark.sql.functions import *
df1= sqlContext.createDataFrame([("81A01","TERR NAME 01","NJ","",""),("81A01","TERR NAME 01","","NY",""),("81A01","TERR NAME 01","","","LA"),("81A02","TERR NAME 01","CA","",""),("81A02","TERR NAME 01","","","NY")], ["zip_code","territory_name","state","state1","state2"])

df1.groupBy("zip_code","territory_name").agg(max("state").alias("state"),max("state1").alias("state1"),max("state2").alias("state2")).show()

Result:

+--------+--------------+-----+------+------+
|zip_code|territory_name|state|state1|state2|
+--------+--------------+-----+------+------+
|   81A02|  TERR NAME 01|   CA|      |    NY|
|   81A01|  TERR NAME 01|   NJ|    NY|    LA|
+--------+--------------+-----+------+------+
user_38155380

Note: For any unique record of zip_code and territory_name, if under any of the state column there are multiple entries, then they would be concatenated.

Some explanation: In this code I employ RDDs. I first divide each record into two tuples, with tuple1 as a key and tuple2 as a value. Then, I reduce by key. x corresponds to tuple1 of (zip_code, territory_name) and tuple2 contains the 3 state columns. tuple1 is taken askey because we want to group by the distinct values of zip_code and territory_name. So, every distinct pair like (81A01,TERR NAME 01) , (81A02,TERR NAME 01) is a key, on the base of which we reduce. Reduce means taking every two values at one time and doing some operation on it and then repeating the same operation with this result and the next element, till the entire tuple in exhausted.

So, reduce of (1,2,3,4,5) with + operation will be - 1+2=3, then 3+3=6 and doing + operation till the last element is reached. Thus, 6+4=10 and finally 10+5=15. Since the tuple ended at 5, so the result is 15. This is how the reduce works with + operation. Since, here we have strings and not numbers, so concatenation will happen A+B=AB.

df1=df1.rdd.map(lambda r: ((r.zip_code, r.territory_name), (r.state, r.state1, r.state2)))\
       .reduceByKey(lambda x,y: (x[0] + y[0], x[1] + y[1], x[2] + y[2]))\
       .map(lambda r: (r[0][0],r[0][1],r[1][0],r[1][1],r[1][2]))\
       .toDF(["zip_code","territory_name","state","state1","state2"])
df1.show()
+--------+--------------+-----+------+------+
|zip_code|territory_name|state|state1|state2|
+--------+--------------+-----+------+------+
|   81A01|  TERR NAME 01|   NJ|    NY|    LA|
|   81A02|  TERR NAME 01|   CA|      |    NY|
+--------+--------------+-----+------+------+
user_83292130
Related