我的 DAG 中有一個 python 運算符。python 可調(diào)用函數(shù)返回一個 bool 值。但是，當我運行 DAG 時，出現(xiàn)以下錯誤。類型錯誤：“bool”對象不可調(diào)用我修改了函數(shù)以不返回任何內(nèi)容，但又一次出現(xiàn)以下錯誤錯誤 - 'NoneType' 對象不可調(diào)用下面是我的狗def check_poke(threshold,sleep_interval):flag=snowflake_poke(1000,10).poke()#print(flag)return flagdependency = 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)無法弄清楚我錯過了什么。有人可以幫我解決這個問題嗎……對氣流相當陌生。我在 dag 中編輯了 python 運算符，如下所示dependency = PythonOperator(task_id='poke_check',provide_context=True,python_callable=check_poke(129600,600),dag=dag)但是現(xiàn)在，我得到了一個不同的錯誤。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)
3 回答
慕標5832272
TA貢獻1966條經(jīng)驗 獲得超4個贊
參數(shù)名稱給出了它。您正在傳遞調(diào)用的結(jié)果而不是可調(diào)用的。
python_callable=check_poke(129600,600)
第二個錯誤指出 callable 是用 25 個參數(shù)調(diào)用的。所以是lambda:行不通的。以下方法可行，但忽略 25 個參數(shù)確實值得懷疑。
python_callable=lambda *args, **kwargs: check_poke(129600,600)
達令說
TA貢獻1821條經(jīng)驗 獲得超6個贊
代碼需要一個可調(diào)用的，而不是結(jié)果（正如已經(jīng)指出的那樣）。
您可以使用functools.Partial來填寫參數(shù)：
from functools import partial
def check_poke(threshold,sleep_interval):
flag=snowflake_poke(1000,10).poke()
return flag
func = partial(check_poke, 129600, 600)
dependency = PythonOperator(
task_id='poke_check',
provide_context=True,
python_callable=func,
dag=dag)
白豬掌柜的
TA貢獻1893條經(jīng)驗 獲得超10個贊
同意@Dan D.的問題；但令人困惑的是為什么他的解決方案不起作用（它在python shell 中肯定有效）
看看這是否會給您帶來任何運氣（它只是@Dan D.解決方案的詳細變體）
from typing import Callable
# your original check_poke function
def check_poke(arg_1: int, arg_2: int) -> bool:
# do something
# somehow returns a bool
return arg_1 < arg_2
# a function that returns a callable, that in turn invokes check_poke
# with the supplied params
def check_poke_wrapper_creator(arg_1: int, arg_2: int) -> Callable[[], bool]:
def check_poke_wrapper() -> bool:
return check_poke(arg_1=arg_1, arg_2=arg_2)
return check_poke_wrapper
..
# usage
python_callable=check_poke_wrapper_creator(129600, 600)
添加回答
舉報
0/150
提交
取消