如何加入 2 个数据框并添加基于过滤器 pyspark 的新列答案

【问题标题】：How to join 2 dataframes and add a new column based on a filter pyspark如何加入 2 个数据框并添加基于过滤器 pyspark 的新列
【发布时间】：2021-11-10 07:13:09
【问题描述】：

我有 2 个数据框，

df1 - cust_id, addr_id,order_time (time at which the order was placed)
df2 - cust_id, addr_id,latest_order (time at which the last order was placed from a specific address per user)

我使用以下命令从 df1 获得了 df2

import pyspark.sql.functions as f

df2 = df1.groupBy('cust_id','addr_id').agg(f.max("order_time").alias('latest_order'))

从这 2 个数据帧中，我想要一个结果数据帧，其中包含为特定 (cust_id,addr_id) 下的订单数，范围从 30 天到最后一次下订单。

样本数据

df1

cust_id     |addr_id        |       order_time  |
+-----------+---------------+-------------------+
|    100    |      1        |2021-01-27         |
|    200    |      2        |2021-01-27         |
|    300    |      3        |2021-01-27         |
|    400    |      4        |2021-01-27         |

df2


+-----------+---------------+-------------------+
|cust_id   |addr_id         |       latest_order|
+-----------+---------------+-------------------+
|   100     |       1       |2021-07-28         |
|   200     |       2       |2021-09-08         |
|   300     |       3       |2020-10-03         |
|   400     |       4       |2020-10-30         |

我尝试使用 UDF，我想将第二个数据帧作为参数传递，但意识到这是不可能的。

所以，我尝试通过以下方式使用 join 代替（不确定语法）

df1.join(df2, ["cust_id","addr_id"], how="inner").select(df1.filter((df1.addr_id == df2.addr_id) & (df1.order_time >= date_sub(df2.latest_order, 30)) & (df1.order_time <= date_sub(df2.latest_order, 1))).count()).alias("order_counts").show()

我最终得到以下错误

org.apache.spark.sql.AnalysisException: Resolved attribute(s) addr_id#433,latest_order#434 missing from cust_id#416,USER_ADDRESS_ID#417,order_time#418 in operator !Filter (((addr_id#417 = addr_id#433) &amp;&amp; (order_time#418 &gt;= cast(date_sub(cast(latest_order#434 as date), 30) as timestamp))) &amp;&amp; (ORDERED_TIME#418 &lt;= cast(date_sub(cast(latest_order#434 as date), 1) as timestamp))). Attribute(s) with the same name appear in the operation: addr_id. Please check if the right attribute(s) are used.;;

我正在学习 PySpark，想知道解决这个问题的正确方法。如果您需要任何其他信息，请告诉我

任何帮助将不胜感激。谢谢！

【问题讨论】：

标签： python dataframe apache-spark join pyspark

【解决方案1】：

通过Window函数可以计算出最新的订单时间，每条记录“order_time”可以与最新的比较：

val df1 = Seq(
  (100, 1, "2021-01-27 13:37:49"),
  (200, 2, "2021-01-27 13:42:02"),
  (300, 3, "2021-01-27 22:20:23"),
  (400, 4, "2021-01-27 22:22:32")
).toDF("cust_id", "addr_id", "order_time")

val customerAddressWindow = Window.partitionBy("cust_id", "addr_id")
val df2 = df1
  .withColumn("order_time", to_timestamp($"order_time", "yyyy-MM-dd HH:mm:ss"))
  .withColumn("latest_order", max("order_time").over(customerAddressWindow))
  .groupBy("cust_id", "addr_id", "latest_order")
  .agg(sum(
    when(datediff($"latest_order", $"order_time") < 30, 1).otherwise(0)
  ).alias("Orders"))

输出是：

+-----------+---------------+-------------------+------+
|cust_id    |addr_id        |latest_order       |Orders|
+-----------+---------------+-------------------+------+
|400        |4              |2021-01-27         |1     |
|300        |3              |2021-01-27         |1     |
|100        |1              |2021-01-27         |1     |
|200        |2              |2021-01-27         |1     |
+-----------+---------------+-------------------+------+

【讨论】：