python - Airflow Python 操作符返回类型

Question

我的 DAG 中有一个 python 运算符。 python 可调用函数返回一个 bool 值。但是，当我运行 DAG 时，出现以下错误。

TypeError: 'bool' object is not callable

我修改了函数以不返回任何内容，但我再次收到以下错误

ERROR - 'NoneType' object is not callable

下面是我的达格

def check_poke(threshold,sleep_interval):
flag=snowflake_poke(1000,10).poke()
#print(flag)
return flag

dependency = PythonOperator(
task_id='poke_check',
#python_callable=check_poke(129600,600),
provide_context=True,
python_callable=check_poke(129600,600),
dag=dag)

end = BatchEndOperator(
queue=QUEUE,
dag=dag)

start.set_downstream(dependency)
dependency.set_downstream(end)

无法弄清楚我错过了什么。有人可以帮我解决这个问题吗...对 Airflow 还很陌生。

我在 dag 中编辑了 python 运算符，如下所示

dependency = PythonOperator(
task_id='poke_check',
provide_context=True,
python_callable=check_poke(129600,600),
dag=dag)

但是现在，我收到了不同的错误。

Traceback (most recent call last):
File "/usr/local/lib/python2.7/dist-packages/airflow/models.py", line 1245, in run
    result = task_copy.execute(context=context)
File "/usr/local/lib/python2.7/dist-packages/airflow/operators/python_operator.py", line 66, in execute
    return_value = self.python_callable(*self.op_args, **self.op_kwargs)
TypeError: () takes no arguments (25 given)
[2019-02-15 05:30:25,375] {models.py:1298} INFO - Marking task as UP_FOR_RETRY
[2019-02-15 05:30:25,393] {models.py:1327} ERROR - () takes no arguments (25 given)

Answer 1

参数名称泄露了它。您正在传递调用的结果而不是可调用的结果。

python_callable=check_poke(129600,600)

第二个错误表明该可调用函数是使用 25 个参数调用的。所以 lambda: 不起作用。以下内容可行，但忽略 25 个参数确实值得怀疑。

python_callable=lambda *args, **kwargs: check_poke(129600,600)