Как преобразовать строку в функцию в Pyspark (Динамическая функция)

Question

Я читаю файл json в Pyspark и динамически генерирую функцию.

Следующая статическая функция была создана перед вызовом файла json (динамическая функция).

def rh_concatenate(*arg):
    return concat(*arg)

def rh_date_to_char(column_name, format1):
    if (format1 == 'MM'):
        return lpad(month(column_name).cast(StringType()), 2, '0')
    elif (format1 == 'YYYY'):
        return lpad(year(column_name).cast(StringType()), 4, '0')

Файл JSON (механизм правил) вернет словарь "exp_list".

словарь exp_list содержит ключ 'QUARTER' и значение 'rh_concatenate (' Q3 ',' - ', rh_date_to_char (' TRANSACTION_DATE ',' YYYY '))'

Сценарий 1: не работает

for key, value in exp_list.items():
    tran_df.withColumn(key, value).show()

Сценарий 2: работает нормально.

tran_df.withColumn('QUARTER', rh_concatenate ('Q3','-',rh_date_to_char ('TRANSACTION_DATE','YYYY' ) )).show()

Я пробовал несколько способов выполнить сценарий 1 и получить приведенную ниже ошибку.

AssertionError: col should be Column

Answer 1

Моя динамическая функция слишком велика, добавлена ​​функция eval, и она отлично работает.

См. Ниже мой ключ и значение.

item key : QUARTER
item value:  rh_if ((rh_less_than ((rh_str_to_num ((rh_date_to_char (('TRANSACTION_DATE'),('MM') )) )),('04') )),(rh_concatenate ((lit('Q1')),(lit('-')),(rh_date_to_char (('TRANSACTION_DATE'),('YYYY') )) )),(rh_if ((rh_less_than ((rh_str_to_num ((rh_date_to_char (('TRANSACTION_DATE'),('MM') )) )),(7) )),(rh_concatenate ((lit('Q2')),(lit('-')),(rh_date_to_char (('TRANSACTION_DATE'),('YYYY') )) )),(rh_if ((rh_less_than ((rh_str_to_num ((rh_date_to_char (('TRANSACTION_DATE'),('MM') )) )),(10) )),(rh_concatenate ((lit('Q3')),(lit('-')),(rh_date_to_char (('TRANSACTION_DATE'),('YYYY') )) )),(rh_concatenate ((lit('Q4')),(lit('-')),(rh_date_to_char (('TRANSACTION_DATE'),('YYYY') )) )) )) )) )

мой код будет выглядеть ниже,

for key, value in exp_list.items():
    print("item key :",key)
    print('item value: ', value)
    if key=='QUARTER':
        tran_df.withColumn(key, eval(value)).show()

Answer 2

Работает нормально ..

Я добавил функцию eval (), и она работает нормально.

tran_df.withColumn(key, eval(value)).show()

Если значение содержит только одну функцию, то вы должны добавить функцию col (),

tran_df.withColumn(key, col(value)).show()

Пример,

def rh_assign(arg):
    return arg

tran_df.withColumn('TRANSACTION_ID', col(rh_assign (('TRANSACTION_ID') ))).show()