如何运行Luigi管道的并行实例:Pid集已在运行

我想用Id 2381启动它一次,然后在第一个作业运行时,我想用Id 231启动第二次运行。第一次运行按预期完成。

Pid(s) set([10362]) already running
Process finished with exit code 0

我是这样开始跑步的

运行一个:

luigi.run(
    cmdline_args=["--id='newId13822'", "--TaskTwo-id=2381"],
    main_task_cls=TaskTwo()
)

运行二:

luigi.run(
    cmdline_args=["--id='newId1322'", "--TaskTwo-id=231"],
    main_task_cls=TaskTwo()
)

每个任务都有一个由luigi的task\u ID\u str(…)方法生成的唯一ID。为什么luigi认为当luigi运行时任务已经在运行。paramater、TaskTwo id和MockTarget文件都不同?

import time
import uuid
from luigi.mock import MockTarget
import luigi


class TaskOne(luigi.Task):
    run_id = luigi.Parameter()

    def output(self):
        return MockTarget("TaskOne{0}".format(self.run_id), mirror_on_stderr=True)

    def run(self):
        _out = self.output().open('w')
        time.sleep(10)
        _out.write(u"Hello World!\n")
        _out.close()


class TaskTwo(luigi.Task):
    id = luigi.Parameter(default=uuid.uuid4().__str__())

    def output(self):
        return MockTarget("TaskTwo{0}".format(self.id), mirror_on_stderr=True)

    def requires(self):
        return TaskOne(self.id)

    def run(self):
        _out = self.output().open('w')
        time.sleep(10)
        _out.write(u"Hello World!\n")
        _out.close()