Luigi: Unexpected behaviour when using luigi.build()

[thisiscab]

Hi,

We've been trying to use the build functionality in order to execute a set of tasks within a task itself. Our use-case is that we want to have some "monitoring" tasks that run after a task has run without adding it has a dependency.

Given the following example:

Class MyS3CopyToTable(redshift.S3CopyToTable, ...):
    def run(self):
        """
        If the target table doesn't exist, self.create_table
        will be called to attempt to create the table.
        """
        if not (self.table):
            raise Exception("table need to be specified")

        path = self.s3_load_path()
        output = self.output()
        connection = output.connect()
        cursor = connection.cursor()

        self.init_copy(connection)
        self.copy(cursor, path)
        self.post_copy(cursor)

        # update marker table
        output.touch(connection)
        connection.commit()

        # commit and clean up
        connection.close()

        # This is what we've added
        luigi.build([myMonitoringTask(params)])

class MyTask(MyS3CopyToTable)
  ...

If we were to run MyTask with one worker (--workers=1), using the central scheduler, then everything executes fine, and my built task (myMonitoringTask) gets executed successfully.

On the other hand, if I were to run it with multiple workers (--workers=2), I see that it spawns my myMonitoringTask correctly but it just hangs. Eventually, it got to state failed with a blank error and leaves my MyTask stuck in the state running and it just hangs our whole pipeline.

There is nothing special in the logs that I can use to determine what the problem would be.
The only thing that happens from time to time is the following:

Task 'myMonitoringTask' is marked as running by disconnected worker 'Worker(salt=xxx, workers=2, host=xxx, username=root, pid=xx)' -> marking as FAILED with retry delay of 1.0s

I'm unsure if we're using the build functionality incorrectly, but given our specific use-case we thought that this would be a good candidate.

Any hints as so why running the above with one worker would work but then if we were to add many workers it would stop working and just "timeout"?

[kwilcox]

I've never seen build used within a run method.

Have you tried yielding the monitoring task instead of using build? Have you tried calling it in an on_success method on the task? Have you tried calling it in a task success notifier?

[thisiscab]

@kwilcox Wow, I'm using yield in many places, it just didn't cross my mind for some kind of voodoo reasons.

All the other suggestions are really good suggestions that I'll play around with and see if they work!

Thank you! :)

[thisiscab]

Alright, I've played with it a little bit more and couldn't get build to work. The worker would eventually fail the task and disconnect itself while leaving the parent task stuck in pending.

In order to remedy for that, I've used a hacky workaround:

    @luigi.Task.event_handler(luigi.Event.SUCCESS)
    def run_monitors(self):
        if self.enable_monitors:
            logger.info('Running monitors')

            sch = scheduler.Scheduler()
            w = worker.Worker(scheduler=sch)

            for task in self.monitor_tasks:
                w.add(task)
            w.run()

This works as expected, but unfortunately, I have to create a brand new scheduler because I can't see to be able to access the current scheduler from a luigi.Task

If anyone has opinions as to how I should do it differently or how I can access the current task scheduler so that I can use that specific scheduler to run a new worker with my task could be great insights!

[kwilcox]

Can you access the scheduler through self (which is actually the task that emitted the SUCCESS event.

@luigi.Task.event_handler(luigi.Event.SUCCESS)
def run_monitors(task):
    w = worker.Worker(scheduler=task.scheduler)
    for mt in task.monitor_tasks:
        w.add(mt)
    w.run()

That being said, yielding the task you want from the run method still seems like a better approach... did you try that?

[thisiscab]

@kwilcox I did try that but that would require me to override all the run() method of every single contrib tasks that we're using, which I don't want to do.

Also, the mixin above has some logic removed from it that adds other functionality that I didn't want to clutter this issue with.

I was pretty certain that I couldn't access the scheduler directly from the task, let me re-validate that.

[erikbern]

I"m a bit confused what you're trying to do here but if the monitoring task should always run after the S3 copy task then you should basically make the s3 copy task a _dependency_ of the monitoring task and let Luigi handle the dependency resolution

[thisiscab]

@erikbern Thanks for raising this question. We might not be using Luigi the conventional way, although I do think there is a lot of similarities. Let's jump right into it

We don't want to have to do something like this:

class Monitor1(luigi.Task):
    required_task = luigi.TaskParameter()

    def run(self):
        ...

    def requires(self):
        return required_task(params)

That would force us to have Monitor1 be aware of the task that needs to run first. Preferably, the monitor shouldn't be aware of what tasks it monitors, but only certain of its parameter.

Also, by going that route, it would imply quite a bit of change on how the tasks interact with each others on our side of things.

We've opted to modify some behaviors of redshift.S3CopyToTable. Let's name it CustomS3CopyToTable that inherits redshift.S3CopyToTable. Our CustomS3CopyToTable overrides some properties like the way it fetches databases configurations and it changes some key options for copy_options.

Then, our tasks inherit CustomS3CopyToTable. For example, we could have something like:

class UpdateTableX(CustomS3CopyToTable):
    ...

   def requires(self): 
        return UpdateTableY(params)

The idea would be to override the run() function of CustomS3CopyToTable and add a yield([monitor1, monitor2]) at the end of the function like such:

def run(self):
        """
        If the target table doesn't exist, self.create_table
        will be called to attempt to create the table.
        """
        if not (self.table):
            raise Exception("table need to be specified")

        path = self.s3_load_path()
        output = self.output()
        connection = output.connect()
        cursor = connection.cursor()

        self.init_copy(connection)
        self.copy(cursor, path)
        self.post_copy(cursor)

        # update marker table
        output.touch(connection)
        connection.commit()

        # commit and clean up
        connection.close()

        yield([monitor1, monitor2])

But that would require us to maintain and make sure that our "custom" run function would match a newer Luigi version if that new version would change the behavior of the original run of S3CopyToTable, like it happens quite often.

kwilcox remembered me that we could use events_handler. For that reason, we've decided to go with a mixins that would add behavior to the specifics contrib that would give this final-ish result:

class MonitorMixin(object):
    @property
    def enable_monitors(self):
        return True

    @property
    def monitor_tasks(self):
        return []

    @luigi.Task.event_handler(luigi.Event.SUCCESS)
    def run_monitors(self):
        if not issubclass(self.__class__, MonitorMixin):
            return

        if self.enable_monitors:
            logger.info('Running monitors')

            sch = scheduler.Scheduler()
            w = worker.Worker(scheduler=sch)

            for task in self.monitor_tasks:
                w.add(task)
            w.run()

class CustomS3CopyToTable(MonitorMixin, redshift.S3CopyToTable):
    ...

class UpdateTableX(DefaultMonitors, CustomS3CopyToTable):
    ...

   def requires(self): 
        return UpdateTableY(params)

class UpdateTableY(DefaultMonitors, CustomS3CopyToTable):
    ...

   def requires(self): 
        return [UpdateTableA(params), UpdateTableB(params), UpdateTableC(params)]

class DefaultMonitors(object):
    def monitor_tasks(self): 
        return [Monitor1(params), Monitor2(params)]

With that mixin, every task that inherits from CustomS3CopyToTable will inherits from MonitorMixin and each of these classes can define their monitors that they want to use.

The above example works as expected and it allows us to do exactly what we want. It's just kinds of hacky because we'd like to be able to call build and forget about the details inside run_monitors.

I now realize that I may have been going into too much details, but I wanted to clarify the purpose of what we wanted to achieve. We want to allow for genericity and do less monkey-patching around what is currently existing so that we keep the default behavior offered by Luigi while adding extra functionality.

I hope that it helps give you a better idea as to where I'm coming from with this! :)

EDIT: I wonder in which use case build is useful if not for the specific use case that I wanted to use it for!

EDIT: We're playing around with different ways to use Luigi and what I've explained in this post may not be what we're going with in the end.

[kwilcox]

I wonder in which use case build is useful if not for the specific use case that I wanted to use it for!

It's very useful when writing tests and calling tasks from notebooks.

We monitor Luigi using the event_handler system. We send chat messages to Flowdock, hit REST endpoints, and produce Kafka messages depending on the Task. These are not Targets because, similar to you, they should not fail the task if they don't run or if they error out. Your Mixin class is similar to our method so I think you'll find success there. The big difference is that our monitoring methods are just functions and not Tasks so we don't need to schedule them.

[erikbern]

Wow thanks for a lot of context @cabouffard !

[thisiscab]

@erikbern My pleasure! :)

@kwilcox Ah! I wouldn't have thought about that type of use-case, for build, that's nice to know!

I'm also happy that you guys went a similar route! I would have liked the idea of having simple functions to log what we wanted to log but in our specific use-case, we look for duplicated keys after our table was loaded and we use a series of tasks that make this happen since we want to save historical data for every step.

Given your scenario, our monitoring tasks query the specific Redshift table and output the data onto S3 then we have a task that parses that file and ensures that we're all good, if we aren't, then we send data to our monitoring system.

By having it in a series of tasks like this, it allows us to simply run the set of tasks if we want to re-run the monitoring for a given table. ANYWAY, all that to say that I think we're good and that my question has been answered.

I'll consider this issue as resolved.

Thank all y'all!