Luigi 工作流程：具有“软”依赖性的任务？答案

【问题标题】：Luigi workflow: tasks with "soft" dependencies?Luigi 工作流程：具有“软”依赖性的任务？
【发布时间】：2019-01-30 22:25:10
【问题描述】：

我正在使用 Luigi 解决依赖关系，效果很好。

我是否也应该将 Luigi 用于“软”依赖项？

让我解释一下。假设我当前的任务下载和处理一堆日期的数据。假设在那之后，我想运行一个脚本来遍历所有数据并输出一个简单的摘要。不是每个日期一个摘要，而是所有已下载数据的一个摘要。

我将此称为软依赖，因为我希望我的最终脚本在所有日期的数据上运行，但如果有几个日期无法下载，我仍然希望脚本在其他日期运行。

我应该如何为这个用例组织任务，或者这不是 Luigi 的工作？

【问题讨论】：

标签： luigi

【解决方案1】：

您只能使用传统的 Luigi 来实现这一点。我可以向您解释如何实现这一点的场景。

其中很少有在 Windows 上无法运行，因此如果您在 Linux 上运行 Luigi，您可以按照以下步骤操作

您需要更改 Luigi 配置。它以 luigi.cfg 文件的形式找到。

[worker]
keep-alive = True

[scheduler]
retry_delay = 10.0 # this is the time the task has to retry. default is 900 seconds. 
retry_count = 3  # how many times the scheduler has to retry before failing it permanently.

retry_delay 值是一个浮点数。这会通知调度程序再次准备失败的任务并将状态从失败更改为挂起。 10秒后，将再次执行。您可以随意更改此值。完成这些更改后，重新启动 Luigi

下面是代码sn-p。我会写一个关于如何使用它的解释

 class FileProcessor(luigi.Task):
      id = luigi.Parameter(default=uuid.uuid4()) # some parameter. Add other parameters as needed here
      counter = 1 # this is important

  #strucuture your output method if you don't need to spit any output in the task. This ensures task completion, without the use of a local target. Local target creates lot of files, which are cumbersome to manage later
      def output(self):
          return luigi.mock.MockTarget(f'FileProcessor-{id}')

      def run(self):
          if self.counter == RETRY_COUNT: #use your retry count here
             log.info(f'Number of retries for this task exceeded {RETRY_COUNT}. Hence soft-failing the task')
             self.output().open('w').close() # this informs the scheduler that the task is complete even though it is not. This is soft failing.

          else:
             log.info(f'Retry count for {self.task_name}is {self.counter}') #just a log for tracking the number of times the task was executed before failing. This is particularly useful when we are hitting APIs with timeout

因此，对于您的问题，您可以使用任一 yield（动态依赖项）来根据日期创建任务。根据日期创建的所有任务都将完成，然后将执行合并（摘要）。即使它们由于某种原因而失败，您也可以使用软失败来执行摘要。

【讨论】：

【解决方案2】：

您可以通过多种方式做到这一点。

内部要求：您可以检查文件是否“存在”，如果存在，则“让出”一个任务来处理该文件。
Inside run：创建一个循环遍历所有文件的巨型任务，如果文件失败，您只需移至列表中的下一项。
Softfail：如果文件已处理，但如果文件不存在，则执行成功的任务。如果文件存在，但在处理过程中发生了一些事情，则使任务失败

优点和缺点：

有利于扩展，因为您可以并行处理这些文件。只要文件的“检查”不会花费太长时间。在 requires 中进行繁重的工作是不可以的，因为它只会延迟整个依赖图的构建。所以如果你有一张大图，那就不好了
实现起来最简单，因为所有处理都在 1 个任务中，输出 1 个摘要文件更容易。
是我的最爱。您不需要在内部进行任何繁重的工作。它可以扩展到大图，您可以并行进行大量处理。这里的一个大问题是它可能会导致那些软故障中的静默错误。如果你能确定这一点，我认为你有一个赢家

【讨论】：

【解决方案3】：

我为这个确切的用例编写了luigi_soft_failures 包。它按照其他两个回复所描述的方式工作，并具有一些额外的选项，例如检查依赖项的软故障、从依赖项传播软故障以及自动捕获错误。

【讨论】：