Implementation:Spotify Luigi Task Lifecycle

Overview

Concrete tool for defining atomic, idempotent units of work with a requires/run/output/complete lifecycle, provided by Luigi.

Description

The luigi.Task class is the central abstraction in Luigi. Every pipeline step is a subclass of Task that overrides up to four key methods:

requires() -- Returns the task(s) that must complete before this task can run. Can return a single Task, a list, or a dict.
run() -- Contains the actual computation. Reads from inputs (provided by dependencies), writes to outputs.
output() -- Returns the Target(s) that this task produces. Used by the framework for completeness checks.
complete() -- Returns True if all outputs exist. The default implementation checks output().exists() for every target returned by output().

An additional convenience method, input(), returns the output() of every task returned by requires(), giving the run() method direct access to upstream data targets.

Usage

Use the Task class when:

You are implementing any pipeline step that reads data, performs computation, and writes results.
You need the Luigi scheduler to track the task's status, manage retries, and prevent redundant execution.
You want to participate in Luigi's dependency resolution and execution graph.

Code Reference

Attribute	Value
Source Location	`luigi/task.py`, lines 150-754
Signature	`class Task(metaclass=Register)`
Import	`from luigi import Task` or `import luigi; luigi.Task`

Key Method Signatures

class Task(metaclass=Register):

    def requires(self):
        """
        The Tasks that this Task depends on.
        A Task will only run if all of the Tasks that it requires are completed.
        Returns a single Task, a list of Task instances, or a dict whose
        values are Task instances.
        """
        return []  # default impl

    def run(self):
        """
        The task run method, to be overridden in a subclass.
        """
        pass  # default impl

    def output(self):
        """
        The output that this Task produces.
        The output of the Task determines if the Task needs to be run --
        the task is considered finished iff the outputs all exist.
        Returns a single Target or a list of Target instances.
        """
        return []  # default impl

    def complete(self):
        """
        If the task has any outputs, return True if all outputs exist.
        Otherwise, return False.
        """
        outputs = flatten(self.output())
        if len(outputs) == 0:
            warnings.warn(
                "Task %r without outputs has no custom complete() method" % self,
                stacklevel=2
            )
            return False
        return all(map(lambda output: output.exists(), outputs))

    def input(self):
        """
        Returns the outputs of the Tasks returned by requires().
        """
        return getpaths(self.requires())

Constructor

def __init__(self, *args, **kwargs):
    params = self.get_params()
    param_values = self.get_param_values(params, args, kwargs)

    for key, value in param_values:
        setattr(self, key, value)

    self.param_kwargs = dict(param_values)
    self._warn_on_wrong_param_types()
    self.task_id = task_id_str(
        self.get_task_family(),
        self.to_str_params(only_significant=True, only_public=True),
    )
    self.__hash = hash(self.task_id)
    self.set_tracking_url = None
    self.set_status_message = None
    self.set_progress_percentage = None

Notable Class Attributes

Attribute	Type	Default	Description
`priority`	`int`	`0`	Higher values are scheduled first.
`disabled`	`bool`	`False`	If `True`, the scheduler marks the task as DISABLED.
`resources`	`dict`	`{}`	Resource requirements, e.g. `{"gpu": 1}`.
`worker_timeout`	`int` or `None`	`None`	Seconds before the run is terminated. 0 or None means no timeout.

I/O Contract

Direction	Name	Type	Description
Input	`requires()` return value	`Task`, `list[Task]`, or `dict[str, Task]`	Upstream tasks whose outputs feed into this task.
Input (derived)	`input()` return value	`Target`, `list[Target]`, or `dict[str, Target]`	The `output()` of each required task, mirroring the structure of `requires()`.
Output	`output()` return value	`Target` or `list[Target]`	The data artifact(s) produced by `run()`.
Output (derived)	`complete()` return value	`bool`	Whether all outputs exist.

Usage Examples

Minimal task with all lifecycle methods

import luigi


class FetchData(luigi.Task):
    date = luigi.DateParameter()

    def output(self):
        return luigi.LocalTarget('/data/raw/%s.json' % self.date)

    def run(self):
        import json
        data = {"date": str(self.date), "values": [1, 2, 3]}
        with self.output().open('w') as f:
            json.dump(data, f)


class TransformData(luigi.Task):
    date = luigi.DateParameter()

    def requires(self):
        return FetchData(date=self.date)

    def output(self):
        return luigi.LocalTarget('/data/processed/%s.csv' % self.date)

    def run(self):
        import json
        with self.input().open('r') as infile:
            data = json.load(infile)

        with self.output().open('w') as outfile:
            outfile.write("date,value\n")
            for v in data['values']:
                outfile.write("%s,%d\n" % (data['date'], v))

Hello World example (from the Luigi repository)

import luigi


class HelloWorldTask(luigi.Task):
    task_namespace = 'examples'

    def run(self):
        print("{task} says: Hello world!".format(task=self.__class__.__name__))


if __name__ == '__main__':
    luigi.run(['examples.HelloWorldTask', '--workers', '1', '--local-scheduler'])

Using complete() and input()

import luigi


class GenerateSummary(luigi.Task):
    date = luigi.DateParameter()

    def requires(self):
        return {
            'sales': SalesData(date=self.date),
            'returns': ReturnsData(date=self.date),
        }

    def output(self):
        return luigi.LocalTarget('/data/summary/%s.txt' % self.date)

    def run(self):
        # input() mirrors the dict structure of requires()
        sales_target = self.input()['sales']
        returns_target = self.input()['returns']

        with sales_target.open('r') as sf, returns_target.open('r') as rf:
            sales_lines = sf.readlines()
            returns_lines = rf.readlines()

        with self.output().open('w') as out:
            out.write("Sales records: %d\n" % len(sales_lines))
            out.write("Return records: %d\n" % len(returns_lines))

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment