在 UDF Pyspark 中更新变量值

Posted 2023-04-13

【中文标题】在 UDF Pyspark 中更新变量值

【英文标题】：Updating Variable Values within UDF Pyspark

【发布时间】：2019-09-16 20:09:41

【问题描述】：

我想要一个通过“值”列的 udf 函数，并检查下一个值是否为当前行值的 50% 或更多。如果它在 50% 以内，那么我想包含值“是”，如果不是，那么我不想包含该值。如果值在最后一个值和下一个值之间下降得太快，则不应包含在内，但如果它逐渐下降且与上一个包含值相比不超过 50%，则没关系。 这就是为什么不包括 id 5 的 .1 而包括 id 9 的 .1 的原因，因为它遵循一个从 0.4 逐渐下降不超过 50% 的值。我正在考虑在 udf 中使用一个变量来跟踪最后一个可接受的值，但我不确定如何去做。

rows = sc.parallelize([[1, .9, 'yes'], [2, .7, 'yes'], [3, .4, 'yes'], [4, .15, 'no'], [5, .1, 'no'], [7, .3, 'yes'], [8, .2, 'yes'], [9, .1, 'yes']])

rows_df = rows.toDF(["ID",  'Values', 'Include'])

#preview data
rows_df.show()

#show data schema
rows_df.printSchema()

+---+------+-------+
| ID|Values|Include|
+---+------+-------+
|  1|   0.9|    yes|
|  2|   0.7|    yes|
|  3|   0.4|    yes|
|  4|  0.15|     no|
|  5|   0.1|     no|
|  7|   0.3|    yes|
|  8|   0.2|    yes|
|  9|   0.1|    yes|
+---+------+-------+

【参考方案1】：

要实现您的目标，您不必使用 UDF（事实上我认为这不可能），您可以使用在窗口上工作的各种功能，例如lag.

我不得不承认我并不完全理解您的要求（为什么 5. 应该是“不”？），但是在 lag、lead 和 last 之间，您应该能够实现它。您可以在the docs 阅读更多关于它们的信息。基于先前值执行逻辑的示例：

from pyspark.sql import Window
from pyspark.sql.functions import col, lag, when, lit

windowSpec = Window.orderBy("Id")

withPrevious = rows_df.withColumn("prevVal", lag(rows_df["Values"]).over(windowSpec))

withPrevious.withColumn("Include2", 
                        when(col("prevVal").isNull(), "yes")\
                        .when(col("Values") >= 0.5 * col("prevVal"), lit("yes"))\
                        .otherwise("no"))\
    .show()

+---+------+-------+-------+--------+
| ID|Values|Include|prevVal|Include2|
+---+------+-------+-------+--------+
|  1|   0.9|    yes|   null|     yes|
|  2|   0.7|    yes|    0.9|     yes|
|  3|   0.4|    yes|    0.7|     yes|
|  4|  0.15|     no|    0.4|      no|
|  5|   0.1|     no|   0.15|     yes|
|  7|   0.3|    yes|    0.1|     yes|
|  8|   0.2|    yes|    0.3|     yes|
|  9|   0.1|    yes|    0.2|     yes|
+---+------+-------+-------+--------+

【讨论】：

我希望第 5 行为否，但第 9 行为是，我想如果无法使用 udf，我将需要使用 for 循环遍历行值。跨度>

您能逐步解释一下如何计算 5 的“否”吗？

在 id1，值为 0.9，id 2 为是，因为 0.7 小于 50% 下降，现在要比较的最后一个接受值是 0.7，所以 id3 是一个是，最后一个接受值现在是 0.4，接下来的两个 id 4-5 是 no，因为它们比 0.4 下降了 50% 以上，然后我们到达 id 7 和 0.3，这是一个是，新接受的值变成 .3，下一个 id也是肯定的，新的 a.v 变为 .2，并且由于 id 9 的 .1 下降不超过 50%，因此在这里被接受