更新UDF Pyspark中的变量值

Question

我想要一个udf函数，它遍历列“Values”，并检查下一个值是否是当前行值的50%或更多。如果它在50%之内，那么我希望包含值“是”，如果不是，那么我不希望包含值。如果该值在最后一个值和下一个值之间下降得太快，则不应将其包括在内，但如果该值逐渐下降，且与最后一个包括的值相比不超过50%，则可以。这就是为什么。未包括id 5的1，但。包含id 9的1，因为它遵循的值从逐渐下降。4不超过50%。我曾考虑在自定义项中使用一个变量来跟踪最后一个可接受的值，但我不知道该怎么做。

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|
+---+------+-------+

Answer 1

为了实现您的目标，您不必使用UDF（事实上我认为这是不可能的），您可以改为使用在窗口上工作的各种函数，例如lag。

我必须承认我不完全理解你的要求（为什么5。应该是‘否’？），但是在滞后、领先和最后之间，你应该能够实现它。你可以在文档中阅读更多关于它们的信息。基于先前值执行逻辑的示例：

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|
+---+------+-------+-------+--------+