我有一个执行一些不稳定计算的 luigi 任务。想想有时不收敛的优化过程。
import luigi
MyOptimizer(luigi.Task):
input_param: luigi.Parameter()
output_filename = luigi.Parameter(default='result.json')
def run(self):
optimize_something(self.input_param, self.output().path)
def output(self):
return luigi.LocalTarget(self.output_filename)
Run Code Online (Sandbox Code Playgroud)
现在我想构建一个包装器任务,该任务将使用不同的输入参数多次运行此优化器,并获取收敛的第一次运行的输出。
我现在实现它的方式是不使用,MyOptimizer因为如果它失败,luigi 会认为包装器任务也失败了,但我可以接受一些MyOptimizer失败的实例。
MyWrapper(luigi.Task):
input_params_list = luigi.ListParameter()
output_filename = luigi.Parameter(default='result.json')
def run(self):
for input_param in self.input_params_list:
try:
optimize_something(self.input_param, self.output().path)
print(f"Optimizer succeeded with input {input_param}")
break
except Exception as e:
print(f"Optimizer failed with input {input_param}. Trying again...")
def output(self):
return luigi.LocalTarget(self.output_filename)
Run Code Online (Sandbox Code Playgroud)
问题在于,通过这种方式,任务不会并行化。此外,您可以想象MyOptimizer和optimize_something参与由 luigi 处理的数据管道的复杂任务,这在我的代码中造成了相当多的混乱。
我将不胜感激有关如何以类似路易吉的方式进行这项工作的任何见解和想法:)
我有一个默认Luigi设置为True的布尔参数的任务:
class MyLuigiTask(luigi.Task):
my_bool_param = luigi.BoolParameter(default=True)
Run Code Online (Sandbox Code Playgroud)
当我从终端运行此任务时,有时我想将该参数作为False传递,但得到以下结果:
$ MyLuigiTask --my_bool_param False
error: unrecognized arguments: False
Run Code Online (Sandbox Code Playgroud)
显然对于false和0...
我知道我可以将默认设置为False,然后--my_bool_param如果我想设置为True,则使用该标志,但我更喜欢使用默认值True。
有什么办法可以做到这一点,并且仍然从终端传递False?