PySpark UDF не распознает количество аргументов

Question

Я определил функцию Python "DateTimeFormat", которая принимает три аргумента

1. Столбец Spark Dataframe с форматами даты (String)
2. Формат ввода значения столбцакак yyyy-mm-dd (String)
3. Формат вывода, т. е. формат, в котором необходимо вернуть ввод, как yyyymmdd (String)

Я сейчас зарегистрированэта функция как UDF в Pyspark. udf_date_time = udf(DateTimeFormat,StringType())

Я пытаюсь вызвать этот UDF в select для фрейма данных, и он, кажется, работает нормально, пока формат ввода и вывода отличаются, как показано ниже

df.select(udf_date_time('entry_date',lit('mmddyyyy'),lit('yyyy-mm-dd')))

Но происходит сбой, если формат ввода и формат вывода совпадают со следующей ошибкой

df.select('exit_date',udf_date_time('exit_date',lit('yyyy-mm-dd'),lit('yyyy-mm-dd')))

«DateTimeFormat» принимает ровно 3 аргумента. 2 дано

Но я явно посылаю три аргумента в UDF

Я пробовал приведенный выше пример на Python 2.7 и Spark 2.1

. Кажется, что функция работает должным образом в обычном Python, когда форматы ввода и вывода совпадают

>>>DateTimeFormat('10152019','mmddyyyy','mmddyyyy')
'10152019'
>>>

Но приведенный ниже код выдает ошибку при запуске в SPARK

import datetime
# Standard date,timestamp formatter
# Takes string date, its format and output format as arguments
# Returns string formatted date
def DateTimeFormat(col,in_frmt,out_frmt):   
    date_formatter ={'yyyy':'%Y','mm':'%m','dd':'%d','HH':'%H','MM':'%M','SS':'%S'}
    for key,value in date_formatter.items():
        in_frmt = in_frmt.replace(key,value)
        out_frmt = out_frmt.replace(key,value)
    return datetime.datetime.strptime(col,in_frmt).strftime(out_frmt)

Вызов UDF с использованиемкод ниже

from pyspark.sql.functions import udf,lit
from pyspark.sql import SparkSession
from pyspark.sql.types import StringType
# Create SPARK session
spark = SparkSession.builder.appName("DateChanger").enableHiveSupport().getOrCreate()

df = spark.read.format("csv").option("header", "true").load(file_path)

# Registering UDF
udf_date_time = udf(DateTimeFormat,StringType())

df.select('exit_date',udf_date_time('exit_date',lit('yyyy-mm-dd'),lit('yyyy-mm-dd'))).show()

Ввод CSV-файла Входной файл

Ожидаемый результат - команда

df.select('exit_date',udf_date_time('exit_date',lit('yyyy-mm-dd'),lit('yyyy-mm-dd'))).show()

НЕ ДОЛЖНА выдавать какую-либо ошибку, такую ​​как DateTimeFormat принимает ровно 3 аргумента, но 2 с учетом

Answer 1

Я не уверен, что есть лучший способ сделать это, но вы можете попробовать следующее:

Здесь я предположил, что вы хотите, чтобы ваши даты были в определенном формате, и установил значение по умолчанию для выходного формата(out_frmt='yyyy-mm-dd') в вашей DateTimeFormat функции

Я добавил новую функцию под названием udf_score, чтобы помочь с преобразованиями. Это может вас заинтересовать

from pyspark.sql.types import StringType
from pyspark.sql.functions import udf, lit


df = spark.createDataFrame([
    ["10-15-2019"],
    ["10-16-2019"],
    ["10-17-2019"],
], ['exit_date'])

import datetime
def DateTimeFormat(col,in_frmt,out_frmt='yyyy-mm-dd'):
    date_formatter ={'yyyy':'%Y','mm':'%m','dd':'%d','HH':'%H','MM':'%M','SS':'%S'}
    for key,value in date_formatter.items():
        in_frmt = in_frmt.replace(key,value)
        out_frmt = out_frmt.replace(key,value)
    return datetime.datetime.strptime(col,in_frmt).strftime(out_frmt)

def udf_score(in_frmt):
    return udf(lambda l: DateTimeFormat(l, in_frmt))

in_frmt = 'mm-dd-yyyy'
df.select('exit_date',udf_score(in_frmt)('exit_date').alias('new_dates')).show()
+----------+----------+
| exit_date| new_dates|
+----------+----------+
|10-15-2019|2019-10-15|
|10-16-2019|2019-10-16|
|10-17-2019|2019-10-17|
+----------+----------+