当在run()中生成任务时，Luigi中的TaskClassAmbigiousException
TaskClassAmbigiousException是Luigi中的一个异常类。当在run()方法中生成任务时，如果出现多个任务类与生成的任务类名称相匹配，就会抛出TaskClassAmbigiousException异常。
Luigi是一个Python模块，用于构建复杂的工作流和任务调度系统。它是一个开源项目，旨在简化数据工程师和科学家在数据处理流程中的工作。Luigi提供了一种声明性的方式来定义任务和任务依赖关系，并提供了可扩展的框架来管理任务的调度和执行。
TaskClassAmbigiousException异常表示在Luigi中存在多个任务类与生成的任务类名称相匹配，这可能是由于命名冲突或代码错误导致的。为了解决这个异常，可以尝试以下几个步骤：
- 检查任务类的命名：确保任务类的名称与其他任务类不冲突，并且在任务类之间有明确的区分。
- 检查导入语句：确保正确导入任务类，并且没有导入重复的任务类。
- 检查任务类的继承关系：如果任务类是从其他类继承而来的，确保没有多个父类具有相同的名称。
如果以上步骤无法解决问题，可能需要进一步检查代码逻辑或查阅Luigi的官方文档和社区支持，以寻求更具体的解决方案。
腾讯云相关产品中与Luigi类似的任务调度和工作流管理工具是Tencent Workflow，它提供了类似的功能，可以用于管理任务和任务依赖关系。您可以通过访问腾讯云的官方网站，了解更多关于Tencent Workflow的详细信息和使用案例。
Tencent Workflow产品介绍链接：https://cloud.tencent.com/product/tencentworkflow
相关·内容
1回答
我是Luigi的新手，正在尝试在我的流程中有一个条件分支。分支任务将评估一个条件，并根据结果跳过它的一些子项。
为了测试这一点，我只有一个虚拟任务，它检查当前小时，如果流在早上执行，则返回True，否则返回False。该任务有两个子任务，一个在控制台中打印'Morning‘，另一个在控制台中打印’下午‘。根据分支任务的结果，一个被激活，另一个被跳过。下面是它在Prefect中的样子：
您可以在这里看到，流是在上午执行的，因此跳过了下午的任务。
在做了一些研究之后，我不知道Luigi是否有能力做这样的事情。到目前为止，我尝试的是：
# Third Task
class Branch
浏览 8提问于2021-06-08得票数 1
1回答
我计划使用Luigi编写一个可重现的、抗故障的超参数调优任务。因此，我在“父”class HParamOptimizer中多次调用我的class TrainOneModel。 为了简单起见，这里是一个更简单的hello world版本： import luigi
# Child class
class HelloTask(luigi.Task):
name = luigi.parameter.Parameter(default='Luigi')
def run(self):
print(f'Luigi says: Hello {se
浏览 18提问于2020-01-09得票数 0
回答已采纳
我正在处理一个Luigi管道，该管道检查手动创建的文件是否存在，如果存在，则继续执行下一个任务：
import luigi, os
class ExternalFileChecker(luigi.ExternalTask):
task_namespace='MyTask'
path = luigi.Parameter()
def output(self):
return luigi.LocalTarget(os.path.join(self.path, 'externalfile.txt'))
class Proce
浏览 2提问于2016-06-23得票数 5
1回答
我是Luigi的新手，我已经创建了一个管道，它从数据库中获取数据，转换数据，然后将其加载回数据库。我在其中创建了四个任务。但是，当我在cmd或Pycharm上执行任务时，它显示不能调度非任务。下面是我的流水线的伪代码。每个任务的参数不是输入，而是从其他文件中获取。 class Task1(luigi.Task):
# Some Parameters
def get_target():
def query():
def run():
class Task2(luigi.Task):
浏览 26提问于2019-08-28得票数 1
回答已采纳
1回答
我意识到我可能需要使用动态需求来完成以下任务，但是我还不能理解这在实践中会是什么样子。 其目标是使用Luigi生成数据并将其添加到数据库中，而无需提前知道将生成什么数据。 以mongodb为例： import luigi
from uuid import uuid4
from luigi.contrib import mongodb
import pymongo
# Make up IDs, though in practice the IDs may be generated from an API
class MakeID(luigi.Task):
def run(self):
浏览 16提问于2019-02-15得票数 0
回答已采纳
我正在与Luigi中的一个我不理解的错误作斗争。我不知道这是一个已知的问题，Luigi的限制还是我做错了什么。 我使用Luigi来解决一个有很多任务和很多依赖项的实际问题。然而，我已经做了一个玩具示例，在这个示例中这个问题看起来很清楚。 让我们考虑两个任务，TaskA和TaskB，TaskA需要执行具有不同Luigi参数值的前两个TaskB实例。 如果我在TaksA的requires()方法中编写依赖项，那么不会有什么不好的事情发生。这三个任务都会执行，并且我会写入退出文件。 但是如果我在TaksA的run()方法中编写依赖项，就会得到丑陋的TaskClassAmbigiousExcepti
浏览 13提问于2021-01-20得票数 2
我在Python中使用luigi，试图理解luigi参数在编译时和运行时是如何工作的。
import luigi
import datetime
class HelloWorld(luigi.Task):
run_dt = luigi.DateSecondParameter(default = datetime.datetime.now())
def requires(self):
return(None)
def output(self):
return(luigi.LocalTarget('helloworld.txt
浏览 2提问于2017-07-18得票数 0
考虑以下任务：
import luigi
class YieldFailTaskInBatches(luigi.Task):
def run(self):
for i in range(5):
yield [
FailTask(i, j)
for j in range(2)
]
class YieldAllFailTasksAtOnce(luigi.Task):
def run(self):
yield [
浏览 37提问于2018-11-28得票数 3
回答已采纳
1回答
我正在使用luigi执行一系列任务，如下所示：
class Task1(luigi.Task):
stuff = luigi.Parameter()
def output(self):
return luigi.LocalTarget('test.json')
def run(self):
with self.output().open('w') as f:
f.write(stuff)
class Task2(luigi.Task):
stuff = luigi.P
浏览 0提问于2018-04-24得票数 5
2回答
我想在Luigi中将一些目标路径设置为全局变量。
原因是我使用的目标路径是基于给定的数值天气预报(NWP)的最后一次运行，并且需要一些时间才能获得该值。一旦我检查了哪一个是最后一次运行，我创建了一个路径，我将在其中放置几个目标文件(具有相同的父文件夹)。
我目前正在重复一个类似的调用来获取几个任务的父路径的值，将此路径设置为全局变量会更有效。我曾尝试在luigi类调用的一个函数(get_target_path)中定义全局变量，但当我返回Luigi管道时，该全局变量似乎无法持久存在。
这就是我的代码的样子：
class GetNWP(luigi.Task):
"""
浏览 0提问于2018-07-03得票数 3
我尝试在名为test.py的命令提示符和名为Test的类中运行1Luigi任务，如下所示
luigi -m test Test --local-scheduler
如何传递参数并在ubuntu中运行
代码示例：
class Test:
def requires(self):
return class1(param1='path', param2='', param3=123)
def run(self):
# some logic with using those 3 params
def out
浏览 0提问于2018-07-24得票数 1
1回答
我需要担心luigi中的文件原子性吗?使用下面的代码，选择一个数据帧并将其作为任务的输出返回。我不明白原子性的部分，因为我希望luigi在声明任务完成之前，只需等待任务完成写入文件。
class readSQLtoPickle(luigi.Task):
sql = luigi.Parameter()
pickle = luigi.Parameter()
def output(self):
return luigi.LocalTarget(self.pickle,format=format.Nop)
def run(self):
浏览 6提问于2020-08-07得票数 0
1回答
我想使用Luigi工作流并行加载spark data frame中的多个文件，并将它们存储在字典中。一旦所有的文件都被加载，我希望能够从main中的字典中访问这些数据帧，然后进一步执行processing.This进程。当我运行Luigi时，一个运行Luigi的worker.if与多个工作人员一起运行Luigi时，此变量在main方法中为空。
任何建议都会有帮助。
import Luigi
from Luigi import LocalTarget
from pyspark import SQLContext
from src.etl.SparkA
浏览 9提问于2021-05-18得票数 1
这是我的的后续，关于循环ETL过程中要遵循的模式。
今天，我写的机器学习工作是手工完成的。我下载所需的输入文件，学习和预测事物，输出一个.csv文件，然后将其复制到数据库中。
然而，由于这将进入生产，我需要自动化所有这些过程。所需的输入文件将每个月(并最终更频繁地)从提供者送到S3桶中。
例如，我试图在Luigi中实现这一点，但是将S3更改为一个本地目录，这样就更简单了。这个程序应该
查看输入目录中的新文件。
当找到新文件时，通过Extract任务将其解压缩到数据目录中。
按Transform任务处理(使用algorithm函数)
它被加载到关于PostgreSQL任务的Loa
浏览 3提问于2017-05-05得票数 0
1回答
我有一个Luigi任务，用于根据传递给它的参数获取不同的数据，另一个任务是接收这些文件并将它们发送到我们的数据仓库。我的问题是，我不知道如何按顺序安排第一个任务的多个版本，以便以后可以将它们发送到datalake。这是我的密码：
class ListStep(luigi.Task):
listId = luigi.IntParameter()
startDate = luigi.Parameter(None)
endDate = luigi.Parameter(None)
logPath = luigi.Parameter(None)
messageP
浏览 3提问于2020-01-07得票数 0
Machine A具有访问SQL数据库的能力，Machine B有访问Google的能力。如果UploadToDrive依赖于DownloadSQLData，那么如何确保任务在正确的机器上运行？
目前，Machine A运行DoSomethingElseWithData，Machine B在几分钟后运行UploadToDrive。这很好，直到有一天Machine A可能无法工作，那时Machine B将尝试将DownloadSQLData作为上游依赖并失败。
class DownloadSQLData(luigi.Task):
# ...
def run(self):
浏览 4提问于2019-10-23得票数 1
回答已采纳
1回答
我试图理解luigi是如何工作的，我理解了这个想法，但是实际的实现要困难一些;)这就是我所拥有的：
class MyTask(luigi.Task):
x = luigi.IntParameter()
def requires(self):
return OtherTask(self.x)
def run(self):
print(self.x)
class OtherTask(luigi.Task):
x = luigi.IntParameter()
def run(self):
y = se
浏览 0提问于2016-09-14得票数 7
回答已采纳
1回答
我有一堆python文件，它们为各种数据源进行端到端的文件处理。
例如，survey.py将使用pandas读取文件，添加列，重命名内容，执行一些计算，然后将修改后的文件保存回磁盘。
对于该文件，driver.py将遵循相同的过程，等等。同样的结构也发生在几十个文件中。
然后我有一个名为process_all的文件，它基本上只是以特定的顺序运行每个python文件(一些文件依赖于其他文件)。在进一步的研究中，我偶然发现了一个名为luigi的库，如果我需要扩展，它似乎可以更有力地完成同样的任务。
问:我是继续编写单独的.py文件来处理数据，还是将所有这些内容都放在luigi的类中？我假设我要处理
浏览 0提问于2016-05-17得票数 2
1回答
今天，我写的机器学习工作是手工完成的。我下载所需的输入文件，学习和预测事物，输出一个.csv文件，然后将其复制到数据库中。
然而，由于这将进入生产，我需要自动化所有这些过程。所需的输入文件将每个月(并最终更频繁地)从提供者送到S3桶中。
现在我计划用Luigi来解决这个问题。以下是理想的过程：
每周(或一天，或每小时，无论我感觉更好)，我都需要我的程序来查看S3桶中的新文件。
当一个文件到达时，我的机器学习管道就会被触发，并吐出一些熊猫的数据。
在此之后，我需要我的程序将这些结果写入不同的DBs中。
问题是，我不知道如何使用Luigi实现自动化：
文件监视
安排任务
浏览 0提问于2017-05-04得票数 1
回答已采纳
我正在使用luigi.build方法尝试luigi的多处理能力。但是我在执行的时候遇到了一些库错误。
for next in self._add(item，is_complete)：文件请求行604，_add self._validate_dependency(d)文件异常行622，_validate_dependency raise Exception('requires() in is_complete objects')
这是我试图实现目标的一段代码。
import luigi
class TaskOne(luigi.Task):
custid= luigi
浏览 16提问于2017-08-11得票数 0
1回答
我很难理解如何在Luigi中实现可重用的任务，然后在具体的情况下使用它们。
例如。我有两个一般任务，对一个文件做一些事情，然后输出结果：
class GffFilter(luigi.Task):
"Filters a GFF file to only one feature"
feature = luigi.Parameter()
out_file = luigi.Parameter()
in_file = luigi.Parameter()
...
class BgZip(luigi.Task):
"bgZips a
浏览 3提问于2017-09-28得票数 0
在luigi中，我正在尝试设置一个工作流，如下所示： 1)解析数据 2)对解析后的数据进行计算 3)将计算出的数据合并在一起 这些操作需要按顺序完成，我有几个工作流程设置如下。然而，即使需求很容易在1和2之间完成(2需要1)，我也不想显式地让3需要2，否则我不能在其他工作流中重用该任务。那么，我该怎么做呢？ 我知道使用动态依赖是可行的，但它的预期用途是在您事先不知道依赖列表的情况下使用，而在这种情况下我知道。它还要求我创建一个按顺序生成任务2和3的Workflow任务，而不仅仅是调度它们。 我尝试过的一种可能的解决方案是创建一个可以将任务作为参数的超类，但不幸的是，这不起作用，因为类不能是参
浏览 15提问于2019-09-03得票数 0
回答已采纳
我有一个测试管道与一个任务一起运行一组luigi任务。我的问题是: luigi任务中的失败不会上升到汇合工作中。换句话说，如果luigi任务失败，completed将不会注册该失败，并声明该汇合作业已成功完成。我将首先发布我正在运行的代码，然后是我尝试过的解决方案。
luigi-tasks.py
class Pipeline1(luigi.WrapperTask):
def requires(self):
yield Task1()
yield Task2()
yield Task3()
tasks.py
class Task1(luig
浏览 18提问于2022-01-10得票数 0
回答已采纳
3回答
我一直在尝试将Luigi整合为我们的工作流程处理程序。目前我们正在使用concourse，但是我们尝试做的许多事情在concourse中是一个麻烦，所以我们切换到Luigi作为我们的依赖管理器。到目前为止，没有任何问题，工作流触发和执行正常。
当一个任务由于某种原因而失败时，问题就出现了。这种情况特别是任务的requires块，但是所有的情况都需要处理。到目前为止，Luigi优雅地处理了这个错误，并将其写入STDOUT。它仍然发出并退出代码0，这意味着作业通过了。假阳性。
我一直在尝试让事件处理来解决这个问题，但是我不能让它触发，即使是一个非常简单的工作：
@luigi.Task.event_
浏览 20提问于2017-03-01得票数 6
1回答
我正在尝试配置Luigi的重试机制，以便将失败的任务重试几次。但是，在成功重新尝试任务时，Luigi未成功退出：
===== Luigi Execution Summary =====
Scheduled 3 tasks of which:
* 2 ran successfully:
- 1 FailOnceThenSucceed(path=/tmp/job-id-18.subtask)
- 1 MasterTask(path=/tmp/job-id-18)
* 1 failed:
- 1 FailOnceThenSucceed(path=/tmp/job-id-18
浏览 2提问于2016-10-24得票数 6
回答已采纳
我在用Luigi发射一些管道。让我们举一个简单的例子
task = myTask()
w = Worker(scheduler=CentralPlannerScheduler(), worker_processes=1)
w.add(task)
w.run()
现在，假设myTask在执行过程中引发异常。我所能得到的只是一个来自luigi的日志，显示了异常。
luigi是否可以传播它，或者至少返回一个failure状态？
这样，我就能够使我的程序对这种状态的功能作出反应。
谢谢。
编辑--我忘了在存储结果时指定luigi的输出正在存储数据库。如果引发异常，则不会存储任何结果，但该异常不会传播给
浏览 1提问于2015-09-15得票数 9
回答已采纳
我想使用Luigi来管理Openstack中的工作流。我对路易吉很陌生。对于初学者，我只想通过Openstack验证自己，然后使用Luigi获取图像列表、风味列表等。任何帮助都是有价值的。
我不太擅长python，但我尝试了下面的代码。我也不能列出图片。错误: glanceclient.exc.HTTPNotFound:找不到资源。(HTTP 404)
import luigi
import os_client_config
import glanceclient.v2.client as glclient
from luigi.mock import MockFile
import sys
浏览 3提问于2017-05-16得票数 1
回答已采纳
1回答
我正在编写一些luigi工作流，并试图调试这些任务。为了做到这一点，我需要用相同的参数一次又一次地重新运行这些任务，直到它们最终完成我想做的事情。
我理解luigi任务是幂等的，因此当给定与前面相同的输入时，通常不会重新运行，这正是之后所需要的--工作流被调试并在生产中运行。然而，在开发过程中，使用完全相同的输入和输出重新运行工作流是有用的--而且，我认为这是必要的。
我知道，在开发期间，我可以重写complete()方法，以在每个任务中返回False。但是，这使任务处于未完成的状态。
我正在寻找一种方法来设置我的工作流在某种“开发”或“调试”模式下运行，这样我就可以一次又一次地运行并重新运行
浏览 4提问于2017-01-23得票数 1
回答已采纳
我有一个requires子任务的Luigi任务。子任务依赖于父任务(即正在执行require执行的任务)传递的参数。我知道您可以通过设置.来指定子任务可以使用的参数。
def requires(self):
return subTask(some_parameter)
...then在子任务上，通过设置.
x = luigi.Parameter()
不过，这似乎只允许您通过一个参数。通过任意数量的参数(任意类型的参数)发送的最佳方法是什么？我真的想要这样的东西：
class parentTask(luigi.Task):
def requires(self):
浏览 1提问于2016-08-24得票数 4
回答已采纳
6回答
目前，我有许多luigi任务一起排队，其中有一个简单的依赖链( a -> b -> c -> d)。首先执行d，最后执行a。a是被触发的任务。
除a之外，所有目标都返回一个luigi.LocalTarget()对象，并有一个泛型luigi.Parameter()，它是一个字符串(包含日期和时间)。在luigi中央服务器上运行(该服务器已启用历史记录)。
问题是，当我重新运行该任务a时，luigi检查历史记录，查看该特定任务以前是否运行过，如果它的状态为“已完成”，它就不会运行任务(在本例中为d)，而且我不能这样做，更改字符串无助于此(添加了一个随机的微秒)。如何强制运行任务？
浏览 11提问于2016-01-05得票数 15
回答已采纳
1回答
我有以下设置
class TaskA(luigi.Task):
def requires(self):
yield TaskB()
if not get_results_from_task_B_written_on_S3():
print('Did not find any results and will exit')
return
else:
print('Found results and will proceed')
yield TaskC
浏览 5提问于2021-11-02得票数 0
2回答
我在我的工作流程中使用Luigi。我的工作流程一般分为三个部分-导入、分析、导出。在每个部分中，都有多个Luigi任务。
我可以把所有的东西都放在一个文件里。但是如果我想把所有东西都分开，就像拥有data_import.py、analysis.py和export.py一样。
例如，如果data_import.py看起来像这样：
import luigi
class import_task_A(luigi.Task):
def requires(self):
return []
def output(self):
return luigi.Loc
浏览 0提问于2016-10-21得票数 5
1回答
前几天，我花了五六个小时在Luigi中尝试并行化一些工作，基于这里使用的方法：
我遇到的问题是，我不断得到一台luigi.task_register.TaskClassAmbigiousException，这让我发疯了。最终，我把luigi.auto_namespace(scope=name)放在包的顶端，一切都开始工作了，但我不知道为什么。粗略地说，我有3项任务：
TaskA -不需要提供带有路径的txt文件
TaskB -只需要输入参数p1，p2提供.csv文件
TaskC -需要任务A的输出，当所有生成的TaskB完成时，输出A的每个路径对生成一个TaskB。
如果有人能勾勒出我该如何正
浏览 0提问于2018-10-26得票数 2
我正在构建我的第一个Luigi管道，并且在构建我的依赖关系之前，我正在单独测试任务。在测试期间，我使用以下主要方法的一个版本来构建一个任务：
if __name__ == "__main__":
headers = dict()
headers["Content-Type"] = "application/json"
headers["Accept"] = "application/json"
luigi.build[(CSVValidator(jsonfile = '/
浏览 0提问于2018-10-25得票数 2
回答已采纳
1回答
第一次进入Luigi (和Python!)还有一些问题。有关守则是：
from Database import Database
import luigi
class bbSanityCheck(luigi.Task):
conn = luigi.Parameter()
date = luigi.Parameter()
def __init__(self, *args, **kwargs):
super(bbSanityCheck, self).__init__(*args, **kwargs)
self.has_run = False
def run(s
浏览 3提问于2014-08-13得票数 2
我已经在luigi框架内实现了一些任务，现在我想要可视化依赖关系图。
我的同事让我开始
luigid mkdir -p luigi;--background --logdir luigi --state-path luigi/state --address 0.0.0.0 --port 32145
之后，我可以在chrome浏览器localhost:32145中看到任务可视化器。
然后我执行luigi任务
python luigi_scheduler.py --local-scheduler
luigi_scheduler按以下行运行该任务
luigi.run(main_task_cls =
浏览 0提问于2014-12-23得票数 4
回答已采纳
以下程序不会输出任何内容，也不会抛出任何错误。我是否遗漏了run()类中的to_S3()方法的形式？
class to_S3(luigi.Task):
#The class Mysql_to_tsv converts the data returned by a query on a Mysqldb and stores the data in a tsv in a local file.
def requires(self):
return [Mysql_to_tsv]
def output(self):
return luigi
浏览 0提问于2015-08-26得票数 1
回答已采纳
1回答
我有一个luigi预处理任务，它将我的原始数据分割成更小的文件。然后，这些文件将由实际管道处理。
因此，关于参数，我想要求每个管道都有一个预处理文件id作为参数。但是，此文件id仅在预处理步骤中生成，因此仅在运行时才知道。为了说明我的想法，我提供了以下不起作用的代码：
import luigi
import subprocess
import random
class GenPipelineFiles(luigi.Task):
input_file = luigi.Parameter()
def requires(self):
pass
de
浏览 1提问于2017-02-21得票数 3
1回答
我已经在Luigi中建立了一个任务管道。由于此管道将在不同的上下文中使用，因此可能需要在管道的开始或结束处包含更多任务，甚至任务之间的依赖关系完全不同。
这时我想：“嘿，为什么要在我的配置文件中声明任务之间的依赖关系呢？”所以我在我的config.py中添加了类似这样的内容：
PIPELINE_DEPENDENCIES = {
"TaskA": [],
"TaskB": ["TaskA"],
"TaskC": ["TaskA"],
"TaskD": ["
浏览 5提问于2017-03-03得票数 8
考虑一个任务通过动态依赖关系依赖于另一个任务的情况： import luigi
from luigi import Task, TaskParameter, IntParameter
class TaskA(Task):
parent = TaskParameter()
arg = IntParameter(default=0)
def requires(self):
return self.parent()
def run(self):
print(f"task A arg = {self.arg}")
c
浏览 25提问于2020-11-15得票数 1
2回答
我有一个读取.sql文件并输出到BigQuery的luigi任务。
我的问题是，有没有方法可以在不同的.sql文件中重用相同的任务，而不必复制整个luigi任务，即我想要创建模板luigi任务的实例。
class run_sql(luigi.task):
sql_file = 'path/to/sql/file' # This is the only bit of code that changes
def complete(self):
...
def requires(self):
...
def run(
浏览 4提问于2018-08-23得票数 1
回答已采纳
我有一个很好的直线工作管道，我在命令行上通过luigi运行的任务触发了所有所需的上游数据的获取和处理，并按正确的顺序进行，直到它滴入我的数据库。
class IMAP_Fetch(luigi.Task):
"""fetch a bunch of email messages with data in them"""
date = luigi.DateParameter()
uid = luigi.Parameter()
…
def output(self):
loc = os.path.join(self.data_dro
浏览 7提问于2017-05-10得票数 2
回答已采纳
2回答
我正在尝试使用Luigi-Jupyter教程链接：
我已经成功地设置了我的环境，但正在为将第一个任务添加到我的管道中而苦苦挣扎。
我使用nano file.py创建了一个python文件，并添加了以下代码：
import os
import luigi
output_path = '/Users/mattiaciollaro/Git/luigi_tutorial/output/'
class TellMeMyName(luigi.Task):
"""
An incredibly simple task that writes your
浏览 3提问于2019-07-24得票数 0
1回答
在我的许多项目中，我使用作为管道工具。这让我想到了使用它来实现参数搜索。标准的luigi.file.LocalTarget对于处理参数有一种非常天真的方法，在文档的示例中也显示了这一点：
def output(self):
return luigi.LocalTarget("data/artist_streams_%s.tsv" % self.date_interval)
即，参数保存在文件名中。这使得检查某个参数组合是否已经计算很容易。一旦任务的参数变得更复杂，这就会变得很混乱。
下面是参数搜索的非常简单的想法：
import luigi
class Sum(luig
浏览 1提问于2016-10-12得票数 2
我是Luigi的新手，我想设置luigi来执行我的API调用。 我正在使用MockFiles，因为我通过API检索的json对象是轻量级的，并且我想避免使用外部数据库。 这是我的代码： import luigi
from luigi import Task, run as runLuigi, mock as LuigiMock
import yaml
class getAllCountries(Task):
task_complete = False
def requires(self):
return LuigiMock.MockFile("a
浏览 36提问于2019-09-06得票数 2
回答已采纳
1回答
这是我第二次尝试理解如何在Luigi中将参数传递给依赖项。第一个是here。 我的想法是:我有依赖于TaskB的TaskC，它依赖于TaskA，它依赖于Task0。我希望整个序列始终是完全相同的，除了我希望能够控制Task0读取的文件，让我们称之为path。Luigi的哲学通常是每个任务应该只知道它所依赖的任务及其参数。这样做的问题是，TaskC、TaskB和TaskA都必须接受变量path，然后将其传递给Task0。 因此，Luigi为此提供的解决方案称为Configuration Classes 下面是一些示例代码： from pathlib import Path
import lui
浏览 10提问于2020-11-23得票数 3
1回答
我最近实现了一个luigi管道来处理我们其中一个生物信息学管道的处理。然而，关于如何设置这些任务，有一些基本的东西我并不了解。
假设我有一个由三个任务组成的链，我希望能够与多个工作者一起运行这些任务。例如，三个工作进程的依赖关系图可能如下所示：
/ taskC -> taskB -> taskA
taskC -> taskB -> taskA
\ taskC -> taskB -> taskA
我可能会写下
class entry(luigi.Task):
in_dir = luigi.Parameter()
def requires(s
浏览 1提问于2017-10-07得票数 1
目前，我正在使用Luigi库来创建数据管道。在我的管道末尾，我有一个Plot函数，它看起来像这样： class PlotAll(luigi.Task):
...
def requires(self):
return{
"tool1" : analyzeTool1Data(...),
"tool2" : analyzeTool2Sata(...)}
def run(self):
data1 = numpy.load(self.input()["tool1
浏览 22提问于2020-08-26得票数 2
回答已采纳
4回答
我试着用一种非常简单的方式来学习路易吉的工作原理。就像一个新手一样，我想出了这个代码
import luigi
class class1(luigi.Task):
def requires(self):
return class2()
def output(self):
return luigi.LocalTarget('class1.txt')
def run(self):
print 'IN class A'
class class2(luigi.Task):
def requires(self):
浏览 12提问于2017-03-30得票数 16
回答已采纳
2回答
Luigi执行方法的顺序是什么(运行、输出、要求)。我理解requires是检查任务DAG有效性的第一次检查，但是不应该在run()之后运行输出吗？
实际上，我正在等待运行中的kafka消息，并在此基础上触发一系列其他任务并返回一个LocalTarget。如下所示：
def run(self):
for message in self.consumer:
self.metadata_key = str(message.value, 'utf-8')
self.path = os.path.join(settings.LUIGI_OUTPU
浏览 3提问于2016-12-29得票数 2
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
