时间戳

首页 > 解决方案 > 从整数列表中选择值的 PySpark 新列

python - 从整数列表中选择值的 PySpark 新列

问题描述

我在 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_Range 代表建筑规范更新的年份，MinYear/MaxYear 代表 ADP 范围的最小和最大年份。

我正在尝试添加具有以下逻辑的最适用的构建代码的列 (ADP_Year)：

预期输出如下：

|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]            |NA      |
|164879|2013   |1999   |2015   |[2003, 2007, 2015, 1999]|2007    |

2010 > MaxYear，所以它从 MaxYear 中选择值，

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

2000 < MinYear 所以它设置为 NA

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

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

dfADP = dfADP.withColumn("ADP_Year",when(dfADP['YearBLT'] < dfADP['MinYear'], lit("NA")\
.when(dfADP['YearBLT'] > dfADP['MaxYear'],dfADP['MaxYear'])))

我正在为此旋转我的轮子，如果这可能的话，我会喜欢一些建议。

标签： pythonapache-sparkpyspark

解决方案

首先让我们找到范围的最大值

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

max_adp_range = array_max("ADP_Range")

和最接近的值：

closest_adp_range = array_max(expr("""
    filter(ADP_Range, y -> y < YearBLT)
"""))

并将这两个组合成一个表达式：

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及filter更高阶函数都需要 Spark 2.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应该更快，但仍然比原生表达式慢）。

推荐阅读