如何根据 pyspark 中的 2 个条件（第一个基于唯一字段，第二个基于日期范围）匹配列？答案

【问题标题】：How to match the column based on 2 conditions (1st based on unique field and 2nd based on date range) in pyspark?如何根据 pyspark 中的 2 个条件（第一个基于唯一字段，第二个基于日期范围）匹配列？
【发布时间】：2022-08-16 12:57:20
【问题描述】：

假设这是我的 1 dataframe 和 userId、deviceID 和 Clean_date（登录日期）

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

假设以下是我的dataframe 2 和userId、deviceID 和transferdate（设备转移到用户ID 的日期）

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

现在，我想确定 3 个场景并使用标识符创建新列

P1 = 用户在同一天使用多台设备登录 df 1，如果两台设备之一不是属于同一用户。
P2 = 用户在 df 1 的不同日期使用多个设备登录，如果两个设备之一不是属于同一用户。

NA = 对于 df 1，用户在同一天/不同天使用多台设备登录，如果两台设备属于同一用户。

因此我的输出表应如下所示：

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

我试过下面的代码：

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\")
    )
)

但是，我无法将 df2 中的 transferdate 合并到此代码中。

任何帮助将不胜感激。

您需要先加入df2，然后才能引用其列。所以，第一个操作应该是：df.join(df2, on=\'userId\').withColumn(...
用户在两个表中是否总是具有相同的 device_id （如您的示例中）？

标签： apache-spark pyspark

【解决方案1】：

如果数据框在 3 列中是唯一的，并且两个表中的用户将拥有相同的设备，则以下解决方案似乎有效。

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|
# +------+--------+----------+-----------+-------------------+-------------------+----------------------------+----------------------------+----------+

【讨论】：