将pyspark偏移滞后动态值检索到其他数据帧

Posted 2023-04-15

【中文标题】将pyspark偏移滞后动态值检索到其他数据帧

【英文标题】：retrieving pyspark offset lag dynamic value in to other dataframe

【发布时间】：2017-11-13 20:04:30

【问题描述】：

我正在使用 pyspark 2.1。下面是我的输入数据框。我坚持从不同的数据帧中获取动态偏移值，请帮助

df1=

类别值

1 3

2 2

4 5

df2

类别年月周数 lag_attribute 运行

1 0 0 0 0 2

1 2019 1 1 1 0

1 2019 1 2 2 0

1 2019 1 3 3 0

1 2019 1 4 4 1

1 2019 1 5 5 2

1 2019 1 6 6 3

1 2019 1 7 7 4

1 2019 1 8 8 5

1 2019 1 9 9 6

2 0 0 0 9 0

2 2018 1 1 2 0

2 2018 1 2 3 2

2 2018 1 3 4 3

2 2018 年 1 3 5 4

如上例所示 df1 是我的查找表，它具有偏移值，对于 1 偏移值是 3 和类别 2 偏移值是 2 。

在 df2 中，runs 是我的输出列，因此对于 df1 中的每个类别值，如果滞后值为 3，则从 dataframe2[df2] 应考虑 lag_attrbute 并滞后 3 个值，因此您可以看到每 3 个值lag_attribute 运行重复

我试过下面的编码没有用。请帮忙

df1=df1.registerTempTable("df1")
df2=df2.registerTempTable("df2")
sqlCtx.sql("select st.category,st.Year,st.Month,st.weekyear,st.lag_attribute,LAG(st.lag_attribute,df1.value, 0) OVER (PARTITION BY st.cagtegory ORDER BY st.Year,st.Month,st.weekyear) as return_test from df1 st,df2 lkp where df1.category=df2.category")

请帮助我跨过这个障碍

【参考方案1】：

lag 接受一个列对象和一个整数（python 整数），如函数签名所示：

Signature: psf.lag(col, count=1, default=None)

count 的值不能是 pyspark IntegerType（列对象）。不过有一些解决方法，让我们从示例数据开始：

df1 = spark.createDataFrame([[1, 3],[2, 2],[4, 5]], ["category", "value"])
df2 = spark.createDataFrame([[1, 0, 0, 0, 0, 2],[1, 2019, 1, 1, 1, 0],[1, 2019, 1, 2, 2, 0],[1, 2019, 1, 3, 3, 0],
                             [1, 2019, 1, 4, 4, 1],[1, 2019, 1, 5, 5, 2],[1, 2019, 1, 6, 6, 3],[1, 2019, 1, 7, 7, 4],
                             [1, 2019, 1, 8, 8, 5],[1, 2019, 1, 9, 9, 6],[2, 0, 0, 0, 9, 0],[2, 2018, 1, 1, 2, 0],
                             [2, 2018, 1, 2, 3, 2],[2, 2018, 1, 3, 4, 3],[2, 2018, 1, 3, 5, 4]], 
                            ["category", "year", "month", "weeknumber", "lag_attribute", "runs"])

如果df1 不太大（意味着少量categories 和每个category 中可能有很多值），你可以做的是转换df1到一个列表并根据其值创建一个 if-elif-elif... 条件：

list1 = df1.collect()
sc.broadcast(list1)

import pyspark.sql.functions as psf
from pyspark.sql import Window
w = Window.partitionBy("category").orderBy("year", "month", "weeknumber")
cond = eval('psf' + ''.join(['.when(df2.category == ' + str(c) + ', psf.lag("lag_attribute", ' + str(l) + ', 0).over(w))' for c, l in list1]))

注意：这是如果c和l是整数，如果它们是字符串那么：

cond = eval('psf' + ''.join(['.when(df2.category == "' + str(c) + '", psf.lag("lag_attribute", "' + str(l) + '", 0).over(w))' for c, l in list1]))

现在我们可以应用条件了：

df2.select("*", cond.alias("return_test")).show()

    +--------+----+-----+----------+-------------+----+-----------+
    |category|year|month|weeknumber|lag_attribute|runs|return_test|
    +--------+----+-----+----------+-------------+----+-----------+
    |       1|   0|    0|         0|            0|   2|          0|
    |       1|2019|    1|         1|            1|   0|          0|
    |       1|2019|    1|         2|            2|   0|          0|
    |       1|2019|    1|         3|            3|   0|          0|
    |       1|2019|    1|         4|            4|   1|          1|
    |       1|2019|    1|         5|            5|   2|          2|
    |       1|2019|    1|         6|            6|   3|          3|
    |       1|2019|    1|         7|            7|   4|          4|
    |       1|2019|    1|         8|            8|   5|          5|
    |       1|2019|    1|         9|            9|   6|          6|
    |       2|   0|    0|         0|            9|   0|          0|
    |       2|2018|    1|         1|            2|   0|          0|
    |       2|2018|    1|         2|            3|   2|          9|
    |       2|2018|    1|         3|            4|   3|          2|
    |       2|2018|    1|         3|            5|   4|          3|
    +--------+----+-----+----------+-------------+----+-----------+

如果df1 是大，那么您可以在已构建的lag 列上自行加入df2：

首先，我们将使用连接将values 从df1 带到df2：

df = df2.join(df1, "category")

如果df1不是太大，你应该broadcast它：

import pyspark.sql.functions as psf
df = df2.join(psf.broadcast(df1), "category")

现在我们将枚举每个partition 中的行并构建一个lag 列：

from pyspark.sql import Window
w = Window.partitionBy("category").orderBy("year", "month", "weeknumber")
left = df.withColumn('rn', psf.row_number().over(w))
right = left.select((left.rn + left.value).alias("rn"), left.lag_attribute.alias("return_test"))

left.join(right, ["category", "rn"], "left")\
    .na.fill(0)\
    .sort("category", "rn").show()

    +--------+---+----+-----+----------+-------------+----+-----+-----------+
    |category| rn|year|month|weeknumber|lag_attribute|runs|value|return_test|
    +--------+---+----+-----+----------+-------------+----+-----+-----------+
    |       1|  1|   0|    0|         0|            0|   2|    3|          0|
    |       1|  2|2019|    1|         1|            1|   0|    3|          0|
    |       1|  3|2019|    1|         2|            2|   0|    3|          0|
    |       1|  4|2019|    1|         3|            3|   0|    3|          0|
    |       1|  5|2019|    1|         4|            4|   1|    3|          1|
    |       1|  6|2019|    1|         5|            5|   2|    3|          2|
    |       1|  7|2019|    1|         6|            6|   3|    3|          3|
    |       1|  8|2019|    1|         7|            7|   4|    3|          4|
    |       1|  9|2019|    1|         8|            8|   5|    3|          5|
    |       1| 10|2019|    1|         9|            9|   6|    3|          6|
    |       2|  1|   0|    0|         0|            9|   0|    2|          0|
    |       2|  2|2018|    1|         1|            2|   0|    2|          0|
    |       2|  3|2018|    1|         2|            3|   2|    2|          9|
    |       2|  4|2018|    1|         3|            4|   3|    2|          2|
    |       2|  5|2018|    1|         3|            5|   4|    2|          3|
    +--------+---+----+-----+----------+-------------+----+-----+-----------+

注意：您的runs 滞后值存在问题，例如，对于catagory=2，它只是滞后1 而不是2。此外，有些行在您的数据框中具有相同的顺序（例如，示例数据框中的最后两行 df2 具有相同的 category, year, month and weeknumber），因为涉及改组，您每次运行代码时可能会得到不同的结果。

【讨论】：

非常感谢 MaFF。我收到 NameError "1" is not defined 。在 eval 函数中 str(c) 的错误。请帮助MaFF

能否把包含在eval中的字符串贴出来（不调用eval函数）

psf.when(df2.category == 1, psf.lag("lag_attribute", 3, 0).over(w)).when(df2.category == 2, psf.lag ("lag_attribute", 2, 0).over(w)).when(df2.category == 4, psf.lag("lag_attribute", 5, 0).over(w))。亲爱的马夫。我已经模拟了数据，但是代替 2 , 4 有字符串值。因此，它会抛出错误，因为名称错误未定义。我假设在与字符串比较时它必须是 df2.category=='abc123' 。但我不知道如何将 str(c) 转换为字符串值。马夫。尝试了不同的方法，比如给 str(c).cast("string") 没有用

非常感谢 MaFF 设法更改数据而不是代码 MaFF