如何使用 on_failure 参数处理 prefect 中的任务失败和 return SUCCESS？

Question

我在 prefect 中有一个 Flow，它的 task 输出是 dataframe。在下面提供的示例中，它总是失败。我想要 task 到 return 一个空的 dataframe，状态为 SUCCESS 使用 @task(on_failure=handle_task_fail)。实现此目的的正确语法是什么？

from pprint import pprint
import pandas as pd

from prefect import Flow, task
from prefect.engine.signals import SUCCESS


def handle_disambig_error(task, old_state, new_state):
    if new_state.is_failed():
        new_state.result["wiki_df"] = pd.DataFrame()

        # Is this needed?
        #set state to SUCCESS
    return new_state


@task(on_failure=handle_disambig_error)
def get_wiki_resource():

    wiki_df = pd.DataFrame(
        {
            "a":[1],
            "b":[1/0]
        }
    )

    return wiki_df

with Flow("Always Fail") as flow:
    wiki_df = get_wiki_resource()

state = flow.run()
task_state = state.result[wiki_df]
pprint(task_state.result)

回溯：

Traceback (most recent call last):
  File "/miniconda3/lib/python3.7/site-packages/prefect/engine/runner.py", line 161, in handle_state_change
    new_state = self.call_runner_target_handlers(old_state, new_state)
  File "/miniconda3/lib/python3.7/site-packages/prefect/engine/task_runner.py", line 120, in call_runner_target_handlers
    new_state = handler(self.task, old_state, new_state) or new_state
  File "/miniconda3/lib/python3.7/site-packages/prefect/utilities/notifications.py", line 69, in state_handler
    fn(obj, new_state)
TypeError: handle_disambig_error() missing 1 required positional argument: 'new_state'
[2020-01-28 17:39:41,759] INFO - prefect.TaskRunner | Task 'get_wiki_resource': finished task run for task with final state: 'Failed'
[2020-01-28 17:39:41,762] INFO - prefect.FlowRunner | Flow run FAILED: some reference tasks failed.

我搜索的一些地方State Handlers, Logging with a State Handler

Answer 1

这里发生了两件事：

1.) 通用状态处理程序：这些可以通过 state_handlers kwarg 设置，并将在每个状态更改时调用。状态处理程序需要具有签名 state_handler(task: Task, old_state: State, new_state: State) -> Optional[State]（这是您正在使用的签名）；调用此处理程序后任务的状态将是从处理程序返回的状态，如果返回 None，则为 new_state。

2.) 失败回调：您在此处使用的 on_failure kwarg 旨在为状态处理程序提供便利 API；传递给此关键字的函数需要具有签名 fn(task: Task, state: State) -> None，并且只会在此任务进入 Failed 状态时调用。请注意，失败回调不能像状态处理程序那样改变任务的状态。

在您的示例中，您似乎混合了两个关键字参数。我相信以下代码会达到您的预期：

from prefect.engine.state import Success


def handle_disambig_error(task, old_state, new_state):
    if new_state.is_failed():
        return_state = Success(result=pd.DataFrame())
    else:
        return_state = new_state
    return return_state

@task(state_handlers=[handle_disambig_error])
def get_wiki_resource():
   return df

如何使用 on_failure 参数处理 prefect 中的任务失败和 return SUCCESS？

How to handle a task failure in prefect and return SUCCESS with the on_failure param?

python

python-3.x

prefect