Spark SQL 场景没有给出正确的结果

Question

我有一个如下的df：

1. 手机号码是different：

   |applicantkey|     first_reg_date|utmcontent| latest_signin_date|mobilenumber|
    +------------+-------------------+----------+-------------------+------------+
    |        1234|2021-01-03 06:05:43|   Android|2021-01-03 06:05:43|         987|
    |        1234|2021-04-03 07:05:43|   Android|2021-10-03 06:05:43|         986|
    +------------+-------------------+----------+-------------------+------------+
   

2. 手机号码是same：

   |applicantkey|     first_reg_date|utmcontent| latest_signin_date|mobilenumber|
    +------------+-------------------+----------+-------------------+------------+
    |        1234|2021-01-03 06:05:43|   Android|2021-01-03 06:05:43|         987|
    |        1234|2021-04-03 07:05:43|   Android|2021-10-03 06:05:43|         987|
    +------------+-------------------+----------+-------------------+------------+
   

现在，我想获取first_reg_date 的min 和latest_signin_date 的max，并替换数据集中这两列的值。所以我的预期输出应该如下所示：

+------------+-------------------+----------+-------------------+------------+
|applicantkey|first_reg_date     |utmcontent|latest_signin_date |mobilenumber|
+------------+-------------------+----------+-------------------+------------+
|1234        |2021-01-03 06:05:43|Android   |2021-10-03 06:05:43|987         |
|1234        |2021-01-03 06:05:43|Android   |2021-10-03 06:05:43|986         |

+------------+-------------------+----------+-------------------+------------+  

我尝试了以下查询，但它给出的输出如下所示：

spark.sql(
    "select applicantkey,min(first_reg_date) first_reg_date,utmcontent,max(latest_signin_date) latest_signin_date,mobilenumber from df group by applicantkey,utmcontent,mobilenumber").show(truncate=False)

+------------+-------------------+----------+-------------------+------------+
|applicantkey|first_reg_date     |utmcontent|latest_signin_date |mobilenumber|
+------------+-------------------+----------+-------------------+------------+
|1234        |2021-01-03 06:05:43|Android   |2021-01-03 06:05:43|987         |
|1234        |2021-04-03 07:05:43|Android   |2021-10-03 06:05:43|986         |
+------------+-------------------+----------+-------------------+------------+  

AND

+------------+-------------------+----------+-------------------+------------+
|applicantkey|first_reg_date     |utmcontent|latest_signin_date |mobilenumber|
+------------+-------------------+----------+-------------------+------------+
|1234        |2021-01-03 06:05:43|Android   |2021-10-03 06:05:43|987         |
+------------+-------------------+----------+-------------------+------------+

第二个输出正确但第一个输出错误。

所以，我尝试了以下方法，它可以帮助我获得正确的结果，但是当手机号码相同时，我会得到重复：

df1 = spark.sql(
    "select applicantkey,min(first_reg_date) first_reg_date, max(latest_signin_date) latest_signin_date from df group by applicantkey")
df2 = spark.sql("select applicantkey,utmcontent,mobilenumber from df")
df3 = df1.join(df2, "applicantkey", "left_outer")
df3.show(truncate=False)  

+------------+-------------------+-------------------+----------+------------+
|applicantkey|first_reg_date     |latest_signin_date |utmcontent|mobilenumber|
+------------+-------------------+-------------------+----------+------------+
|1234        |2021-01-03 06:05:43|2021-10-03 06:05:43|Android   |987         |
|1234        |2021-01-03 06:05:43|2021-10-03 06:05:43|Android   |987         |
+------------+-------------------+-------------------+----------+------------+  

我不想在最后使用DISTINCT()。那么，我到底做错了什么？

Answer 1

根据您的评论，您希望根据 applicantkey 为数据框中的所有行选择 min(first_reg_date) 和 max(latest_signin_date)。您可以通过window functions 执行此操作。 我们将通过applicantkey 对数据框进行分区，对于此分区中的每一行，我们将填充min(first_reg_date) 和max(latest_signin_date) 的值；我们将保持其他列不变。

最小工作示例

from datetime import datetime

from pyspark.sql.types import *
import pyspark.sql.functions as F
from pyspark.sql import Window

data = [(1234,datetime.strptime("2021-01-03 06:05:43", "%Y-%m-%d %H:%M:%S"),"Android",datetime.strptime("2021-01-03 06:05:43", "%Y-%m-%d %H:%M:%S"),987),
         (1234,datetime.strptime("2021-04-03 07:05:43", "%Y-%m-%d %H:%M:%S"),"Android",datetime.strptime("2021-10-03 06:05:43", "%Y-%m-%d %H:%M:%S"),986),
         (1234,datetime.strptime("2021-01-03 06:05:43", "%Y-%m-%d %H:%M:%S"),"Android",datetime.strptime("2021-01-03 06:05:43", "%Y-%m-%d %H:%M:%S"),987),
         (1234,datetime.strptime("2021-04-03 07:05:43", "%Y-%m-%d %H:%M:%S"),"Android",datetime.strptime("2021-01-03 06:05:43", "%Y-%m-%d %H:%M:%S"),987),
  ]

schema = StructType([
    StructField("applicantkey",IntegerType(),True),
    StructField("first_reg_date",TimestampType(),True),
    StructField("utmcontent",StringType(),True),
    StructField("latest_signin_date", TimestampType(), True),
    StructField("mobilenumber", IntegerType(), True),
  ])
 
df = spark.createDataFrame(data=data,schema=schema)

窗口逻辑

window_spec = Window.partitionBy("applicantkey")
df.withColumn("first_reg_date", F.min(F.col("first_reg_date")).over(window_spec)).withColumn("latest_signin_date", F.max(F.col("latest_signin_date")).over(window_spec)).show()

输出

+------------+-------------------+----------+-------------------+------------+
|applicantkey|     first_reg_date|utmcontent| latest_signin_date|mobilenumber|
+------------+-------------------+----------+-------------------+------------+
|        1234|2021-01-03 06:05:43|   Android|2021-10-03 06:05:43|         987|
|        1234|2021-01-03 06:05:43|   Android|2021-10-03 06:05:43|         986|
|        1234|2021-01-03 06:05:43|   Android|2021-10-03 06:05:43|         987|
|        1234|2021-01-03 06:05:43|   Android|2021-10-03 06:05:43|         987|
+------------+-------------------+----------+-------------------+------------+

PS。将来在提问时，请务必包含一个最低限度的工作示例。