如何在 Pyspark 中的一列日期中减去一列天数

Question

给定以下 PySpark DataFrame

df = sqlContext.createDataFrame([('2015-01-15', 10),
                                 ('2015-02-15', 5)],
                                 ('date_col', 'days_col'))

如何从日期列中减去天数列？在此示例中，生成的列应['2015-01-05', '2015-02-10'] 。

我研究了pyspark.sql.functions.date_sub()，但它需要一个日期列和一个一天，即 date_sub(df['date_col'], 10) .理想情况下，我宁愿做date_sub(df['date_col'], df['days_col']).

我还尝试创建一个 UDF：

from datetime import timedelta
def subtract_date(start_date, days_to_subtract):
    return start_date - timedelta(days_to_subtract)
subtract_date_udf = udf(subtract_date, DateType())
df.withColumn('subtracted_dates', subtract_date_udf(df['date_col'], df['days_col'])

这在技术上是可行的，但我已经读到在Spark和Python之间跳跃可能会导致大型数据集的性能问题。我现在可以坚持使用这个解决方案（不需要过早优化），但我的直觉告诉我，必须有一种方法可以在不使用 Python UDF 的情况下完成这个简单的事情。

Answer 1

使用expr函数（如果您有从列中减去的dynamic values）：

>>> from pyspark.sql.functions import *
>>> df.withColumn('substracted_dates',expr("date_sub(date_col,days_col)"))

使用与列函数（如果您有literal values要减去）：

>>> df.withColumn('substracted_dates',date_sub('date_col',<int_literal_value>))

Answer 2

我能够使用 selectExpr 解决这个问题。

df.selectExpr('date_sub(date_col, day_col) as subtracted_dates')

如果要将列追加到原始 DF，只需将*添加到表达式中

df.selectExpr('*', 'date_sub(date_col, day_col) as subtracted_dates')

Answer 3

不是有史以来最优雅的解决方案，但如果你不想在 Scala 中破解 SQL 表达式（并不是说它应该很难，但这些是sql私有的），这样的事情应该可以解决问题：

from pyspark.sql import Column
def date_sub_(c1: Column, c2: Column) -> Column:
    return ((c1.cast("timestamp").cast("long") - 60 * 60 * 24 * c2)
        .cast("timestamp").cast("date"))

对于Python 2.x，只需删除类型注释即可。

Answer 4

格式略有不同，但也可以：

df.registerTempTable("dfTbl")
newdf = spark.sql("""
                     SELECT *, date_sub(d.date_col, d.day_col) AS DateSub 
                     FROM dfTbl d
                   """)