是否可以在 Luigi 管道的两个不同位置重用任务？答案

【问题标题】：Is it possible to reuse a task in two different spots in a Luigi pipeline?是否可以在 Luigi 管道的两个不同位置重用任务？
【发布时间】：2017-11-15 23:47:10
【问题描述】：

我正在开发 Luigi 管道，但遇到了一个重大问题。似乎没有任何方法可以在同一管道中重用任务。为了说明我所追求的，请考虑以下工作流程：

Collect_Data → Clean_data → Task_on_data_A → Task_on_data_B ↳ Sample_data → Task_on_data_A → Task_on_data_B

我对@987654323@ 和Sample_data 执行相同的操作。我想在遵循 DRY 原则的同时在 Luigi 中做这样的事情。现在看来我必须从字面上复制并粘贴任务 A 和 B 的代码才能做到这一点。

Airflow 有 a way to deal with situations like this. Luigi 吗？

【问题讨论】：

标签： python luigi

【解决方案1】：

我认为您应该将箭头绘制为相反的依赖箭头。以这种方式考虑它会有所帮助，因为执行路径是由 requires 方法确定的。在您的场景中，这非常简单，因为您只有一个具有条件依赖的 TaskA。

您只需要在开始时再添加一个依赖于两个 TaskB 的任务和一个指示该 taskb 是否要在样本上运行的指示符

伪代码：

ParentTask
  def requires():
     return [TaskB(sample_type=True), 
             TaskB(sample_type=False)]

TaskB
  def requires():
    return TaskA(sample_type)

TaskA
  def requires():
    if sample_type:
      return SampleTask()
    else:
      return CleanTask()

SampleTask
  def requires():
    return CleanTask()

CleanTask
  def requires()
    return CollectData()

【讨论】：

毫无疑问这是可行的，但它不会使代码重用变得复杂吗？我需要在我的任务中对数据源进行硬编码。因此，如果有人稍后来制作不同的 sample_data 并想要使用 TaskA-TaskB，他们必须编辑源（或重新创建）TaskA？感觉很反鸭子打字。 python 中的某些东西感觉很奇怪
Andrew- 我不建议在任务中对数据源进行硬编码。最好在 ParentTask 中对其进行参数化并将其传递到堆栈中。但这是一种不同的思维方式，因为您必须始终从您想要的结果开始，并考虑这会如何影响您实现目标所需的步骤，而不是从考虑第一步开始。