【问题标题】：Pyspark: Feature Engineer Point-in-Time MetricsPyspark：特征工程师时间点指标
【发布时间】：2020-11-02 23:44:42
【问题描述】：

我一直在尝试对时间点数据建模。为了举例说明这个过程，假设有 4 位不同的客户（一些是回头客，一些是新客户）从商店购物（如下所示）

我正在尝试使用 Pyspark 按 ID 划分我的代码，并创建两个特征：平均（在那个时间点）购买和总购买。结果如下：

我已经通过简单地进行累积求和确定了 Total_Purchase_1 列，但我无法在我的一生中获得仅使用该日期的滞后值计算平均值的时间点平均值 (Mean_Purchase_1)。

有没有人有过处理此类特征工程任务的经验？我是一个大 R 人，并且有使用 dplyr 甚至 Python 中的 Pandas 处理它的经验，但我正在尝试使用 Pyspark 优化它。谢谢！

【问题讨论】：

您想如何计算平均购买量？如果那是一组客户？即，对于 B55 - 它总是 500，如果是这种情况，我不明白其他两个值是如何出现的 - 平均购买 1 的第二和第三行的 650 和 100 .. 如果你能帮助澄清逻辑
嘿@dsk 感谢您的提问。这些指标基于该日期的当前平均值。例如，2015 年 10 月 2 日是客户 B55 的第一次购买，所以他们的平均值等于他们的购买 (1000/1) 2016 年 1 月 5 日是他们的第二次购买（300 美元），所以在那个时间点，他们的平均值是 (1000+300)/2 = 650
抱歉延迟回复..我已经分享了一个解决方案，如果这对您有用，请您检查并帮助接受和投票..

标签： python pyspark apache-spark-sql feature-engineering

【解决方案1】：

这应该是适合您的解决方案，请使用 window function 和 mean()

在此处创建 DF

from pyspark.sql import functions as F
from pyspark.sql import types as T
from pyspark.sql.window import Window as W
df = spark.createDataFrame([("B55","2018-05-28", 200),
                           ("B55","2016-05-01", 300),
                           ("B55","2015-02-10", 1000),
                            ("A37","2017-12-30", 2100),
                            ("A37","2016-06-21", 2000)
                           ],[ "id","date","value"])
df.show()
_w = W.partitionBy("id").orderBy("date")
df = df.withColumn("mean", F.mean("value").over(_w))
df.show()

输入

+---+----------+-----+
| id|      date|value|
+---+----------+-----+
|B55|2018-05-28|  200|
|B55|2016-05-01|  300|
|B55|2015-02-10| 1000|
|A37|2017-12-30| 2100|
|A37|2016-06-21| 2000|
+---+----------+-----+

输出

+---+----------+-----+------+
| id|      date|value|  mean|
+---+----------+-----+------+
|A37|2016-06-21| 2000|2000.0|
|A37|2017-12-30| 2100|2050.0|
|B55|2015-02-10| 1000|1000.0|
|B55|2016-05-01|  300| 650.0|
|B55|2018-05-28|  200| 500.0|
+---+----------+-----+------+

【讨论】：

很高兴看到它对您有所帮助 :) 您是否愿意为答案投赞成票 :) 将不胜感激！