Conditionnal branch in Luigi Workflows

Viewed 257

I am new to Luigi and trying to have a conditional branch in my flow. The branch task would evaluate a condition, and depending on the result some of its children would be skipped.

For testing this, I just have a dummy task that checks the current hour and returns True if the flow is executed during the morning, False otherwise. This task has two children, one that prints 'Morning' in the console, and one that prints 'Afternoon'. Depending on the result of the branching task, one is activated and the other is skipped. Here is what it looks like in Prefect: Prefect version

You can see here that the flow was executed during the morning, so the afternoon task was skipped.

After doing some research, I don't know if Luigi is capable of doing something like this or not. What I tried so far is this:

# Third Task
class Branch(luigi.Task):
    def requires(self):
        return Sleep()

    def output(self):
        return luigi.LocalTarget('condition.txt')

    def run(self):
        date = str(datetime.now())
        hour = int (date.split()[1].split(':')[0])
        with self.output().open('w') as out:
            if hour<12:
                out.write('morning')
                open("afternoon.txt", "w").close() # create afternoon Target skips afternoon task
            else:
                out.write('afternoon')
                open("morning.txt", "w").close() # create morning Target skips afternoon task

# Fourth Task depends on result of branch
class Morning(luigi.Task):
    def requires(self):
        return Branch()

    def output(self):
        return luigi.LocalTarget('morning.txt')

    def run(self):
        print ("Morning")
        with self.output().open('w') as out:
            out.write("Morning")


class Afternoon(luigi.Task):
    def requires(self):
        return Branch()

    def output(self):
        return luigi.LocalTarget('afternoon.txt')

    def run(self):
        print ("Afternoon")
        with self.output().open('w') as out:
            out.write("Afternoon")

# Fifth task is a merge after the branch
class Merge(luigi.WrapperTask):
    def requires(self):
        yield Morning()
        yield Afternoon()

    def run(self):
        print ("Merged")

From what I understood, Luigi will only execute a task if its output does not exist yet. My idea was then to create the output file of the task to skip to prevent it from executing, but it does not work.

1 Answers

Luigi has the feature of dynamical dependencies. In this case, you can have the Merge task require the Branch task. While the Merge task runs, depending on the output of the Branch task, it can dynamically yield either the Morning or the Afternoon task.

Related