sciluigi 0.9.2b3

Note: this library is still work in progress, but it has already been put into production and has been executing real-world workloads successfully, and no serious flaws are known. Early-adopters are much welcome to test it out and file issues for anything not working properly, features missing, or suggestions for improvements!

Scientific Luigi (SciLuigi for short) is a light-weight wrapper library around Spotify’s Luigi workflow system that aims to make writing scientific workflows (consisting of numerous interdependent commandline applications) more fluent, flexible and modular.

Luigi is a great, flexible, and very fun-to-use library. It has turned out though, that its default way of defining dependencies by hard coding them in each task’s requires() function is not optimal for some type of workflows common in scientific fields such as bioinformatics, where multiple inputs and outputs, complex dependencies, and the need to quickly try different workflow connectivity (e.g. plugging in extra filtering steps) in an explorative fashion is central to the way of working.

SciLuigi was designed to solve some these problem we were facing when trying to use luigi for defining complex workflows for data preprocessing, machine-learning and cross-validation.

To achieve that, SciLuigi provides the following “features” over vanilla Luigi:

• Separation of dependency definitions from the tasks themselves, for improved modularity and composability.

• Inputs and outputs implemented as separate fields, a.k.a. “ports”, to allow specifying dependencies between specific input and output-targets rather than just between tasks. This is again to let such details of the network definition reside outside the tasks.

• The fact that inputs and outputs are object fields, also allows auto-completion support to ease the network connection work (Works great e.g. with jedi-vim.

• Inputs and outputs are connected with an intuitive “single-assignment syntax”.

• Good default high-level logging of workflow tasks and execution times.

• Produces an easy to read audit-report with high level information per task.

• Integration with some HPC workload managers. (So far only SLURM though).

Because of Luigi’s great easy-to-use API, these changes have been implemented as a very thin layer on top of luigi’s own API, and no changes to the luigi core is needed at all, so you can continue leveraging the work already being put into maintaining and further developing luigi, by the team at Spotify and others.

import sciluigi as sl

class MyWorkflow(sl.WorkflowTask):
    def workflow(self):
        # Initialize tasks:
        foowrt = self.new_task('foowriter', MyFooWriter)
        foorpl = self.new_task('fooreplacer', MyFooReplacer,
            replacement='bar')

        # Here we do the *magic*: Connecting outputs to inputs:
        foorpl.in_foo = foowrt.out_foo

        # Return the last task(s) in the workflow chain.
        return foorpl

import sciluigi

class MyWorkflow(sciluigi.WorkflowTask):
    def workflow(self):
        pass # TODO: Implement workflow here later!

class MyFooWriter(sciluigi.Task):
    # We have no inputs here
    # Define outputs:
    def out_foo(self):
        return sciluigi.TargetInfo(self, 'foo.txt')
    def run(self):
        with self.out_foo().open('w') as foofile:
            foofile.write('foo\n')

class MyFooReplacer(sciluigi.Task):
    replacement = luigi.Parameter() # Here, we take as a parameter
                                  # what to replace foo with.
    # Here we have one input, a "foo file":
    in_foo = None
    # ... and an output, a "bar file":
    def out_replaced(self):
        # As the path to the returned target(info), we
        # use the path of the foo file:
        return TargetInfo(self, self.in_foo().path + '.bar.txt')
    def run(self):
        with self.in_foo().open() as in_f:
            with self.out_replaced('w') as out_f:
                # Here we see that we use the parameter self.replacement:
                out_f.write(in_f.read().replace('foo', self.replacement))

def run(self):
    # Here, we use the in-built self.ex() method, to execute commands:
    self.ex("sed 's/foo/{repl}' {in} > {out}".format(
        repl=self.replacement,
        in=self.in_foo().path,
        out=self.out_bar().path))

import sciluigi
class MyWorkflow(sciluigi.WorkflowTask):
    def workflow(self):
        foowriter = self.new_task('foowriter', MyFooWriter)
        fooreplacer = self.new_task('fooreplacer', MyFooReplacer,
            replacement='bar')

        # Here we do the *magic*: Connecting outputs to inputs:
        fooreplacer.in_foo = foowriter.out_foo

        # Return the last task(s) in the workflow chain.
        return fooreplacer

# End of script ....
if __name__ == '__main__':
    sciluigi.run_local(main_task_cls=MyWorkflow)

python myworkflow.py