0

我有以下 pyspark df 可以通过代码重新创建

df = spark.createDataFrame([(1, "John Doe", "2020-11-30"),(2, "John Doe", "2020-11-27"),(3, "John Doe", "2020-11-29")],
                            ("id", "name", "date")) 

   +---+--------+----------+
| id|    name|      date|
+---+--------+----------+
|  1|John Doe|2020-11-30|
|  2|John Doe|2020-11-27|
|  3|John Doe|2020-11-29|
+---+--------+----------+

我希望创建一个 udf 来计算 2 行日期之间的差异（使用 Lag 函数），不包括周末，因为 pyspark 2.2.0 没有内置函数来执行此操作。例如。2020-11-30 和 2020-11-27 之间的差异应该是 1，因为它们分别是星期一和星期五。

我试图在 python 中计算两个日期之间的差异（不包括周末）的帮助下创建以下内容：

from pyspark.sql.functions import udf
import numpy as np
workdaUDF = udf(lambda z: workdays(z),IntegerType())
def workdays():
date1 = df.select(F.col('date')).collect()[1][0]
date2 = df.select(F.col('date')).collect()[0][0]
date_diff = np.busday_count(date1,date2)
return date_diff

df.withColumn("date_dif",workdaysUDF(F.col("date"))).show(truncate=False)

但我收到以下错误

PicklingError: Could not serialize object: Exception: It appears that you are attempting to reference SparkContext from a broadcast variable, action, or transformation. SparkContext can only be used on the driver, not in code that it run on workers. For more information, see SPARK-5063.

关于如何在我的数据框的每一行上进行这项工作的任何帮助都会非常有帮助。

PS：我的 date1 和 date2 变量需要是动态的，具体取决于应用函数的日期值。此外，由于数据框的大小，我无法使用我找到多种解决方案的 pandas。

先感谢您。

4

1 回答 1

2

您不能调用collectUDF。您只能将列传递给UDF，因此您应该传入日期列和lag日期列，如下所示：

import numpy as np
import pyspark.sql.functions as F
from pyspark.sql.window import Window
from pyspark.sql.types import IntegerType

df = spark.createDataFrame([
    (1, "John Doe", "2020-11-30"),
    (2, "John Doe", "2020-11-27"),
    (3, "John Doe", "2020-11-29")],
    ("id", "name", "date")
) 

workdaysUDF = F.udf(lambda date1, date2: int(np.busday_count(date2, date1)) if (date1 is not None and date2 is not None) else None, IntegerType())
df = df.withColumn("date_dif", workdaysUDF(F.col('date'), F.lag(F.col('date')).over(Window.partitionBy('name').orderBy('id'))))
df.show()

+---+--------+----------+--------+
| id|    name|      date|date_dif|
+---+--------+----------+--------+
|  1|John Doe|2020-11-30|    null|
|  2|John Doe|2020-11-27|      -1|
|  3|John Doe|2020-11-29|       1|
+---+--------+----------+--------+
于 2020-12-01T07:52:19.853 回答