Функция Pyspark UDF выдает ошибку - PullRequest
Новая
       35

Функция Pyspark UDF выдает ошибку

1 голос
/ 05 ноября 2019

Я пытаюсь добиться разницы между двумя значениями столбца timestamp. Попытка достичь того же результата, используя различные методы, доступные в спарк. Я могу достичь того же результата, используя Spark SQL и обычную функцию. Но когда я пытаюсь зарегистрировать эту функцию как UDF, она начинает выдавать ошибку.

Данные: 

id|end_date|start_date|location
1|2015-10-14 00:00:00|2015-09-14 00:00:00|CA-SF
2|2015-10-15 01:00:20|2015-08-14 00:00:00|CA-SD
3|2015-10-16 02:30:00|2015-01-14 00:00:00|NY-NY
4|2015-10-17 03:00:20|2015-02-14 00:00:00|NY-NY
5|2015-10-18 04:30:00|2014-04-14 00:00:00|CA-SD

Использование SparkSQL: отлично работает !! 

data.createOrReplaceTempView("data_tbl")
query = "SELECT id, end_date, start_date,\
        datediff(end_date,start_date) as dtdiff FROM data_tbl"

spark.sql(query).show()

Использование функции Python: работает отлично !! 

from pyspark.sql.functions import datediff

def get_diff(x, y):
    result = datediff(x,y)
    return result

data.withColumn('differ',get_diff('end_date','start_date')).show()

результат в обоих случаях: 

+---+-------------------+-------------------+--------+------+
| id|           end_date|         start_date|location|differ|
+---+-------------------+-------------------+--------+------+
|  1|2015-10-14 00:00:00|2015-09-14 00:00:00|   CA-SF|    30|
|  2|2015-10-15 01:00:20|2015-08-14 00:00:00|   CA-SD|    62|
|  3|2015-10-16 02:30:00|2015-01-14 00:00:00|   NY-NY|   275|
|  4|2015-10-17 03:00:20|2015-02-14 00:00:00|   NY-NY|   245|
|  5|2015-10-18 04:30:00|2014-04-14 00:00:00|   CA-SD|   552|
+---+-------------------+-------------------+--------+------+

Регистрация функции как UDF: не работает !! 

from pyspark.sql.functions import udf, datediff
get_diff_udf = udf(lambda x, y: datediff(x,y))
data.withColumn('differ',get_diff_udf('end_date','start_date')).show()

Ошибка:

Py4JJavaError: An error occurred while calling o934.showString.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 18.0 failed 1 times, most recent failure: Lost task 0.0 in stage 18.0 (TID 18, localhost, executor driver): org.apache.spark.SparkException: Python worker failed to connect back.

1 Ответ

2 голосов
/ 05 ноября 2019

Вам необходимо отключить безопасность вилки, установив для переменной окружения OBJC_DISABLE_INITIALIZE_FORK_SAFETY значение YES. Это решило ту же проблему для меня.

Вы можете включить это в свой скрипт: 

import os
os.environ['OBJC_DISABLE_INITIALIZE_FORK_SAFETY'] = 'YES'

Чтобы узнать больше о fork safety или почему нам нужно установить эту переменную env:

Многопроцессорная обработка приводит к сбою Python и выдает ошибку, возможно, происходившую в другом потоке при вызове fork ()

...