但是如果我在TaksA的run()方法中编写依赖项，就会得到丑陋的TaskClassAmbigiousException。首先，生成requieres()中的任务，它可以工作： import luigi
j = luigi.IntParameter(defaultLuigi说"NoneType“
浏览 13提问于2021-01-20得票数 2
1回答
前几天，我花了五六个小时在Luigi中尝试并行化一些工作，基于这里使用的方法：TaskA -不需要提供带有路径的txt文件
TaskB -只需要
浏览 0提问于2018-10-26得票数 2
考虑以下任务：
def run(self):
for i inYieldFailTaskInBatches在for循环中多次生成FailTask，而YieldAllFailTasksAtOnce生成数组中的所有任务。如果我运行YieldFailTaskInBatches，Luigi
浏览 37提问于2018-11-28得票数 3
回答已采纳
2回答
我开始使用Luigi了。我已经构建了一个执行多个任务的管道，并且我非常小心地确保这些任务正常工作。所以管道运转良好。在构建管道时，在任务发生故障的时候，他们会在:(中得到报告，我会编辑它们，直到它们正常工作为止。我希望我的管道不会在出现故障时停止，而是跳过(也许还会记录故障)，然后继续下一种情况。(这些“失败”并不是因为任务写得不好，而是因为我们假设输入到这些任务的</em
浏览 1提问于2020-10-29得票数 0
6回答
目前，我有许多luigi任务一起排队，其中有一个简单的依赖链( a -> b -> c -> d)。首先执行d，最后执行a。a是被触发的任务。除a之外，所有目标都返回一个luigi.LocalTarget()对象，并有一个泛型luigi.Parameter()，它是一个字符串(包含日期和时间)。在luigi中央服务器上运行(该服务器已启用历史记录)。问题是，当我重新运行该任务a时，luigi
浏览 11提问于2016-01-05得票数 15
回答已采纳
我有一个测试管道与一个任务一起运行一组luigi任务。我的问题是: luigi任务中的失败不会上升到汇合工作中。换句话说，如果luigi任务失败，completed将不会注册该失败，并声明该汇合作业已成功完成。我将首先发布我正在运行的代码，然后是我尝试过的解决方案。如果需要以特定的方式/位置生成错误，我就这样做了。虽然
浏览 18提问于2022-01-10得票数 0
回答已采纳
2回答
我想在Luigi中将一些目标路径设置为全局变量。我目前正在重复一个类似的调用来获取几个任务的父路径的值，将此路径设置为全局变量会更有效。我曾尝试在luigi类调用的一个函数(get_target_path)<em
浏览 0提问于2018-07-03得票数 3
我有一个任务，它生成哪些文件应该被处理： def run(self): def output(self):我
浏览 3提问于2018-01-24得票数 2
回答已采纳
1回答
我意识到我可能需要使用动态需求来完成以下任务，但是我还不能理解这在实践中会是什么样子。 其目标是使用Luigi生成数据并将其添加到数据库中，而无需提前知道将生成什么数据。up IDs, though in practice the IDs may be generated from an API def runvalid_ids,
浏览 16提问于2019-02-15得票数 0
回答已采纳
我开始将一个夜间数据管道从一个可视化ETL工具移植到Luigi，我非常喜欢有一个可视化工具来查看作业的状态。但是，我注意到，在上一个作业(名为MasterEnd)完成几分钟后，除了MasterEnd之外，所有节点都从图形中消失。这有点不方便，因为我想看到每一天/过去几天的一切都完成了。需要注意的一点是，我有一个cron，它每15分钟运行一次这个管道，以检查S3上的文件。如果它存在，它就会运行，否则就会停止。我不确定这是否会导致任务从可视化器中
浏览 1提问于2015-09-17得票数 6
回答已采纳
2回答
我在我的工作流程中使用Luigi。我的工作流程一般分为三个部分-导入、分析、导出。在每个部分中，都有多个Luigi任务。 def output(self): def run(self):my import stuff
if __name__ == '__ma
浏览 0提问于2016-10-21得票数 5
1回答
我正在使用芹菜作为我的网络应用程序。芹菜执行父任务，然后执行更多的任务。芹菜问题
这会有什么问题吗?我
浏览 1提问于2016-05-05得票数 3
回答已采纳
我正在编写一个流水线，在这个流水线中，后续任务需要读取早期任务的输出，这样它们就可以知道需要在请求中传递哪些参数。 number = luigi.IntParameter()
pass
yield SquareAll()
yield
浏览 1提问于2017-06-05得票数 3
3回答
我一直在尝试将Luigi整合为我们的工作流程处理程序。目前我们正在使用concourse，但是我们尝试做的许多事情在concourse中是一个麻烦，所以我们切换到Luigi作为我们的依赖管理器。当一个任务由于某种原因而失败时，问题就出现了。这种情况特别是任务的requires块，但是所有的情况都需要处理。到目前为止，Luigi优雅地处理了这个错误，并将其写入STDOUT。def output(sel
浏览 20提问于2017-03-01得票数 6
2回答
我有一个读取.sql文件并输出到BigQuery的luigi任务。我的问题是，有没有方法可以在不同的.sql文件中重用相同的任务，而不必复制整个luigi任务，即我想要创建模板luigi任务的实例。class run_sql(luigi.task):
sql_file = 'path/to/
浏览 4提问于2018-08-23得票数 1
回答已采纳
1回答
我正在使用luigi执行一系列任务，如下所示： stuff = luigi.Parameter()
with self.output().open('w(stuff)
当我像这样启动整个工作流时，这完全可以正常
浏览 0提问于2018-04-24得票数 5
在luigi中，我正在尝试设置一个工作流，如下所示： 1)解析数据 2)对解析后的数据进行计算 3)将计算出的数据合并在一起 这些操作需要按顺序完成，我有几个工作流程设置如下。然而，即使需求很容易在1和2之间完成(2需要1)，我也不想显式地让3需要2，否则我不能在其他工作流中重用该任务。那么，我该怎么做呢？我知道使用动态依赖是可行的，但它的预期用途是在您事先不知道依赖列表的情况下使用，而在这种情况下我知道。它还要求我创建一个按顺序<e
浏览 15提问于2019-09-03得票数 0
回答已采纳
1回答
我有一个luigi预处理任务，它将我的原始数据分割成更小的文件。然后，这些文件将由实际管道处理。 yield RunPipelineOnSmallChunk(directory=self.input_file, file_id=file_id)
luigi.run其次，要求实际的</em
浏览 1提问于2017-02-21得票数 3
我正在处理一个Luigi管道，该管道检查手动创建的文件是否存在，如果存在，则继续执行下一个任务： luigi.run(['MyTask.ProcessExternalFile','--path', path,\
'-
浏览 2提问于2016-06-23得票数 5
我熟悉Luigi事件处理机制，并且我已经实现了一个管道，当管道成功完成时，会发送一个成功的电子邮件。我想知道当任务已经运行，因此任务不再重新运行时，如何触发事件。举个例子-当一个新的日期文件按每日计划出现时，我的作业就会被触发。周日，没有新文件出现，luigi生成以下输出：* 1 p
浏览 0提问于2018-02-13得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
腾讯会议
云直播
对象存储活动推荐
腾讯云开发者
