How to match the column based on 2 conditions (1st based on unique field and 2nd based on date range) in pyspark?

user_16363630 0
Viewed 49

Suppose this is my 1 dataframe with userId, deviceID and Clean_date (date of log in)

df =

userId deviceID Clean_date
ABC123 202030 28-Jul-22
XYZ123 304050 27-Jul-22
ABC123 405032 28-Jul-22
PQR123 385625 22-Jun-22
PQR123 465728 22-Jun-22
XYZ123 935452 22-Mar-22

Suppose following is my dataframe 2 with userId, deviceID and transferdate (date of device transferred to userid)

df2 =

userId deviceID transferdate
ABC123 202030 20-May-22
XYZ123 304050 03-May-22
ABC123 405032 02-Feb-22
PQR123 385625 21-Jun-22
PQR123 465728 2-Jul-22
XYZ123 935452 26-Apr-22

Now, I want to identify 3 scenarios and create new column with identifier

  1. P1 = User logging in with multiple devices on same day for df 1 and if one of the both devices are not belonging the same user.
  2. P2 = User logging in with multiple devices on different day for df 1 and if one of the both devices are not belonging the same user.
  3. NA = User logging in with multiple devices on same day/different day for df 1 and if both devices are belonging the same user.

Hence my output table should look like:

df3 =

userId deviceID Clean_date transferdate identifier
ABC123 202030 28-Jul-22 20-May-22 NA
XYZ123 304050 27-Jul-22 03-May-22 P2
ABC123 405032 28-Jul-22 02-Feb-22 NA
PQR123 385625 22-Jun-22 21-Jun-22 P1
PQR123 465728 22-Jun-22 02-Jul-22 P1
XYZ123 935452 22-Mar-22 26-Apr-22 P2

I have tried below code:

from pyspark.sql import functions as f, Window

w=Window.partitionBy("userId") 
w2 = Window.partitionBy("userId", "Clean_date") 
df3 = (
    df
    .withColumn(
        "Priority",
        f.when(f.size(f.collect_set("deviceID").over(w2)) > 1, "P1")
        .when(f.size(f.collect_set("deviceID").over(w)) > 1, "P2")
        .otherwise("NA")
    )
)

However, I am unable to incorporate transferdate from df2 in this code.

Any help would be greatly appreciated.

pyspark apache-spark
1 Answers

If the dataframes are unique at the 3 columns and the users in both tables will have same devices, the below solution seems to work.

data1_sdf.join(data2_sdf, ['userid', 'deviceid'], 'left'). \
    withColumn('num_dev_sameday_gt1', 
               (func.count('deviceid').over(wd.partitionBy('userid', 'clean_dt')) > 1).cast('int')
               ). \
    withColumn('num_dev_diffday_gt1', 
               (func.size(func.collect_set('clean_dt').over(wd.partitionBy('userid'))) > 1).cast('int')
               ). \
    withColumn('sameday_atleast_1dev_notuser', 
               func.max(((func.col('num_dev_sameday_gt1') == 1) & (func.col('clean_dt') < func.col('transfer_dt'))).cast('int')).
               over(wd.partitionBy('userid'))
               ). \
    withColumn('diffday_atleast_1dev_notuser', 
               func.max(((func.col('num_dev_diffday_gt1') == 1) & (func.col('clean_dt') < func.col('transfer_dt'))).cast('int')).
               over(wd.partitionBy('userid'))
               ). \
    withColumn('identifier',
               func.when((func.col('num_dev_sameday_gt1') == 1) & (func.col('sameday_atleast_1dev_notuser') == 1), func.lit('P1')).
               when((func.col('num_dev_diffday_gt1') == 1) & (func.col('diffday_atleast_1dev_notuser') == 1), func.lit('P2')).
               otherwise(func.lit('NA'))
               ). \
    show()

# +------+--------+----------+-----------+-------------------+-------------------+----------------------------+----------------------------+----------+
# |userid|deviceid|  clean_dt|transfer_dt|num_dev_sameday_gt1|num_dev_diffday_gt1|sameday_atleast_1dev_notuser|diffday_atleast_1dev_notuser|identifier|
# +------+--------+----------+-----------+-------------------+-------------------+----------------------------+----------------------------+----------+
# |PQR123|  385625|2022-06-22| 2022-06-21|                  1|                  0|                           1|                           0|        P1|
# |PQR123|  465728|2022-06-22| 2022-07-02|                  1|                  0|                           1|                           0|        P1|
# |XYZ123|  304050|2022-07-27| 2022-05-03|                  0|                  1|                           0|                           1|        P2|
# |XYZ123|  935452|2022-03-22| 2022-04-26|                  0|                  1|                           0|                           1|        P2|
# |ABC123|  202030|2022-07-28| 2022-05-20|                  1|                  0|                           0|                           0|        NA|
# |ABC123|  405032|2022-07-28| 2022-02-02|                  1|                  0|                           0|                           0|        NA|
# +------+--------+----------+-----------+-------------------+-------------------+----------------------------+----------------------------+----------+
user_82795850
Related