从整数列表中选择值的新列

Question

我在PySpark中有以下数据帧：

|ID    |YearBLT|MinYear|MaxYear|ADP_Range               |
---------------------------------------------------------
|164876|2010   |2004   |2009   |[2004,2009]             |
|164877|2008   |2000   |2011   |[2000, 2002, 2011]      |
|164878|2000   |2003   |2011   |[2003, 2011]            |
|164879|2013   |1999   |2015   |[2003, 2007, 2015, 1999]|

其中YearBLT是房地产的建造年份，ADP_范围表示建筑代码更新的年份，MinYear/MaxYear表示ADP范围的最小和最大年份。在

我正在尝试添加一列（ADP峎Year），该列具有以下逻辑的最适用建筑代码：

• 如果YearBLT小于MinYear，则ADP_Year==“NA”
• 如果YearBLT大于MaxYear，则ADP_Year==Max（ADP_范围）
• 如果YearBLT介于两者之间，则选择ADP_范围内YearBLT下方的最近日期

预期产出如下：

^{pr2}$

2010>；MaxYear，因此它从MaxYear中选择值

2008年在2000年和2011年之间；因为有第三个值是2002年，所以选择这个值是因为它比2000年更近

2000<；MinYear所以它设置为NA

2013年在1999年和2015年之间；因为第三和第四个值是2007年和2015年，所以选择2007年

前两种情况很简单，我有它们的工作代码：

^{3}$

我正在这方面旋转我的车轮，如果这甚至是可能的话，我希望有一些建议。在

Answer 1

首先让我们找出最大范围

from pyspark.sql.functions import array_max, col, expr, when

max_adp_range = array_max("ADP_Range")

最接近的值：

^{pr2}$

把这两个结合成一个表达式：

adp_year = when(
    # If the YearBLT is greater than the MaxYear, ADP_Year == Max(ADP_Range)
    col("YearBLT") > col("MaxYear"), max_adp_range
).when(
    # If the YearBLT is in between, it chooses 
    # the closest date below the YearBLT in the ADP_Range
    col("YearBLT").between(col("MinYear"), col("MaxYear")), closest_adp_range
).otherwise(
   # If the YearBLT is less than the MinYear, ADP_Year == "NA"
   # Note: not required. Included just for clarity.
   None
)

最后选择：

df = spark.createDataFrame([                                    
    (164876, 2010, 2004, 2009, [2004,2009]),
    (164877, 2008, 2000, 2011, [2000, 2002, 2011]),   
    (164878, 2000, 2003, 2011, [2003, 2011]),         
    (164879, 2013, 1999, 2015, [2003, 2007, 2015, 1999])
], ("id", "YearBLT", "MinYear", "MaxYear", "ADP_Range"))

df.withColumn("ADP_YEAR", adp_year).show()

这将产生预期结果：

+   +   -+   -+   -+          +    +
|    id|YearBLT|MinYear|MaxYear|           ADP_Range|ADP_YEAR|
+   +   -+   -+   -+          +    +
|164876|   2010|   2004|   2009|        [2004, 2009]|    2009|
|164877|   2008|   2000|   2011|  [2000, 2002, 2011]|    2002|
|164878|   2000|   2003|   2011|        [2003, 2011]|    null|
|164879|   2013|   1999|   2015|[2003, 2007, 2015...|    2007|
+   +   -+   -+   -+          +    +

array_max和{}高阶函数都需要spark2.4或更高版本。在2.3或之前，您可以将以上表达式重新定义为

from pyspark.sql.functions import udf

max_adp_range = udf(max, "bigint")("ADP_Range")
closest_adp_range = udf(
    lambda xs, y: max(x for x in xs if x < y), "bigint"
)("ADP_Range", "YearBLT")

但是您应该预期会有显著的性能损失（单个udf应该更快，但仍然比本机表达式慢）。在