Можем ли мы использовать ключевые аргументы в UDF - PullRequest
Новая
       21

Можем ли мы использовать ключевые аргументы в UDF

0 голосов
/ 08 мая 2018

Вопрос, который у меня возникает, - можем ли мы использовать ключевые аргументы вместе с UDF в Pyspark, как я это сделал ниже. Метод conv имеет ключевое слово аргумент conv_type, который по умолчанию назначается определенному типу форматера, однако в некоторых местах я хочу указать другой формат. Который не проходит через udf из-за ключевого аргумента. Есть ли другой подход использования ключевого аргумента здесь? 

from datetime import datetime as dt, timedelta as td,date

tpid_date_dict = {'69': '%d/%m/%Y', '62': '%Y/%m/%d', '70201': '%m/%d/%y', '66': '%d.%m.%Y', '11': '%d-%m-%Y', '65': '%Y-%m-%d'}

def date_formatter_based_on_id(column, date_format):
    val = dt.strptime(str(column),'%Y-%m-%d').strftime(date_format)
    return val

def generic_date_formatter(column, date_format):
    val = dt.strptime(str(column),date_format).strftime('%Y-%m-%d')
    return val

def conv(column, id, conv_type=date_formatter_based_on_id):
    try:
        date_format=tpid_date_dict[id]
    except KeyError as e:
        print("Key value not found!")
    val = None
    if column:
        try:
            val = conv_type(column, date_format)
        except Exception as err:
            val = column
    return val

conv_func = functions.udf(conv, StringType())

date_formatted = renamed_cols.withColumn("check_in_std", 
conv_func(functions.col("check_in"), functions.col("id"), 
generic_date_formatter))

Так что проблема в последнем утверждении ( date_formatted = renamed_cols.withColumn ("check_in_std", conv_func (functions.col ("check_in"), functions.col ("id"), generic_date_formatter)) ) Поскольку третий аргумент generic_date_formatter является аргументом ключевого слова.

При попытке этого я получаю следующую ошибку: AttributeError: у объекта 'function' нет атрибута '_get_object_id'

1 Ответ

0 голосов
/ 08 мая 2018

К сожалению, вы не можете использовать udf с аргументами ключевых слов. UserDefinedFunction.__call__ определяется только с позиционными аргументами : 

def __call__(self, *cols):
    judf = self._judf
    sc = SparkContext._active_spark_context
    return Column(judf.apply(_to_seq(sc, cols, _to_java_column)))

но ваша проблема не связана с аргументами ключевых слов. Вы получаете исключение, потому что generic_date_formatter это не Column объект, а функция.

Вы можете создать udf динамически: 

def conv(conv_type=date_formatter_based_on_id):
    def _(column, id):
        try:
            date_format=tpid_date_dict[id]
        except KeyError as e:
            print("Key value not found!")
        val = None
        if column:
            try:
                val = conv_type(column, date_format)
            except Exception as err:
                val = column
        return val
    return udf(_, StringType())

который можно назвать: 

conv_func(generic_date_formatter)(functions.col("check_in"), functions.col("id"))

Проверка Передача столбца фрейма данных и внешнего списка в udf в withColumn для получения подробной информации.

...