在flink1.17.1中java调用pythonUDF奇怪问题[maximum recursion depth][python函数正常]

Thu, 10 Aug 2023 07:10:48 -0700 

HI
我简化了我python代码
只要udft方法有外部方法,都会有递归问题, 比如 
agan_add_iig(),尽管我的agan_add_iig()实现很简单,flink难道不能外部import自定义方法吗??
def agan_add_iig():
return 2
@udtf(input_types=DataTypes.STRING(),
result_types=[DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING(), 
DataTypes.STRING()])
def run(data_str):
try:
        logger.info("input param is %s", data_str)
data = [{'name': data_str}]
        start_time = time.time()

        agan_add_iig()
        end_time = time.time()

print("入参耗时:", end_time - start_time)
# extractEngine = init_info(data)
return 'success', 'success', 'success', 'success'

except Exception as e:
        err = e
        logger.error(e)
return str(err), '', '', ''

回复