在其他Dataframe中检索pyspark offset lag动态值

我正在使用PySpark2.1。下面是我的输入Dataframe。我很喜欢从不同的Dataframe中获取动态偏移值请帮助
df1型=
类别值
1 3
2 2
4 5
df2型
类别年月周数lag\u属性运行
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中的每个类别值，如果lag值是3，那么从dataframe2[df2]应该考虑lag\u attrbute和lag down的3个值，因此可以看到每3个lag\u属性值都重复运行
我试过了，但没用。请帮忙

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

请帮我跨过这个障碍

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

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

的值 count 不可能是火种 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 大了就可以自己加入了 df2 建立在一个 lag 列：
首先我们要把 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 例如。还有一些行具有相同的顺序（例如，示例Dataframe中的最后两行） df2 有相同的 category, year, month and weeknumber )在您的Dataframe中，由于涉及洗牌，所以每次运行代码时可能会得到不同的结果。

在其他Dataframe中检索pyspark offset lag动态值

1条答案

相关问题

热门标签

最新问答