问题标签 [luigi]

我正在使用 django 作为 Web 框架。我需要一个可以执行同步和异步（批处理任务）任务链的工作流引擎。我发现 celery 和 luigi 作为批处理工作流程。我的第一个问题是这两个模块之间有什么区别。

Luigi 允许我们重新运行失败的任务链，并且只有失败的子任务才能重新执行。celery 呢：如果我们重新运行链（在修复失败的子任务代码之后），它会重新运行已经成功的子任务吗？

假设我有两个子任务。第一个创建一些文件，第二个读取这些文件。当我将这些放入芹菜链中时，由于第二个任务中的错误代码，整个链失败。当我在第二个任务中修复代码后重新运行链时会发生什么？第一个任务会尝试重新创建这些文件吗？

第一次进入 Luigi（和 Python！）领域并有一些问题。相关代码为：

输出在这里（删除了相关数据库返回的原因）：

所以问题：

1.) 为什么“我即将屈服”会打印两次？

2.) 为什么从来没有打印出“hello world”？

3.) 什么是“1 个可能由其他工作人员运行的待处理任务”？

我更喜欢超级干净的输出，因为它更容易维护。我希望我能消除这些警告等价物。

我还注意到需要“产量”或“返回项目、项目 2、项目 3”。我读过关于产量并理解它。我没有得到的是哪种约定在这里被认为是优越的，或者它们是否是我对语言的新手没有得到的细微差异。

在 Web 界面和https://github.com/spotify/luigi/blob/master/luigi/task.py中，我可以看到一个任务可以有“资源”。在一个名为 process_resources() 的 Task 类中还有一个占位符函数，它只返回作为资源的空字典。

这个神秘的资源是什么东西？

据我所知， aluigi.Target可以存在，也可以不存在。因此，如果 aluigi.Target存在，则不会重新计算。

我正在寻找一种方法来强制重新计算任务，如果它的一个依赖项被修改，或者如果其中一个任务的代码发生变化。

我需要使用来自 python的Luigi运行 Hadoop jar 作业。我搜索并找到了在 Luigi 中编写 mapper 和 reducer 的示例，但没有直接运行 Hadoop jar。

我需要运行直接编译的 Hadoop jar。我该怎么做？

我正在使用 Spotify 的 Luigi 工具来处理多个作业之间的依赖关系。

在我的示例中，我想要求 from task1，然后检索一些取决于执行的信息task1，以便将其作为 的参数传递task2。但是，我的功能retrieve_info将无法工作，因为task1尚未运行。

我的问题是，既然我正在使用yield，task1在调用 of 之前不应该处理retrieve_info吗？Luigi 是否在迭代所需的功能，然后启动不同任务的处理？

如果最后一个假设是正确的，我如何使用所需任务的执行作为第二个所需类的输入？

我正在尝试模拟为 luigi 参数提供默认值的东西。

一个愚蠢的例子，展示了我想要完成的事情：

待测任务：

单元测试代码：

luigi.Parameter 逻辑似乎比补丁更早发生。

在此示例中，test_patched_foo通过和test_patched_baz失败。所以补丁确实发生了，但发生在luigi.Parameter(default=bar.bar())线路调用之后。

是否可以模拟和修补以这种方式调用的东西？

我只是尝试从文档中运行 python luigi示例：

我使用命令行运行它：

我的印象是这会在我运行它的目录中创建一个文件，但它没有？

难道我做错了什么？

我有通过触发 Luigi

在我的课堂上，我有：

但是任务测试似乎无法解析我正确输入的参数？

我正进入（状态：

Luigi 是否支持我执行整个流程？

流程是这样的：将 MySQL 数据转储到 S3，然后使用 Copy 命令将数据移动到 Redshift。

我可以使用 Luigi 执行上述工作流程吗？