Luigi

https://github.com/spotify/luigi
Admin
Luigi is a Python package that helps you build complex pipelines of batch jobs, handling dependency...
Tokens:32,076
Snippets:293
Trust Score:9.2
License:Apache-2.0
Update:1 month ago
Show doc for...
Tokens:
Raw
### Install uv and tox

Source: https://github.com/spotify/luigi/blob/master/CONTRIBUTING.rst

Commands to install the uv package manager and the tox test runner.

```bash
curl -LsSf https://astral.sh/uv/install.sh | sh
uv tool install tox --with tox-uv
```

--------------------------------

### Example TOML configuration

Source: https://github.com/spotify/luigi/blob/master/doc/configuration.rst

A sample configuration file using the TOML format, which can be enabled by setting LUIGI_CONFIG_PARSER=toml.

```python
[hadoop]
version = "cdh4"
streaming_jar = "/usr/lib/hadoop-xyz/hadoop-streaming-xyz-123.jar"

[core]
scheduler_host = "luigi-host.mycompany.foo"
```

--------------------------------

### Start Central Scheduler

Source: https://github.com/spotify/luigi/blob/master/doc/example_top_artists.rst

Command to launch the Luigi central scheduler server, which requires Tornado.

```console
$ luigid
```

--------------------------------

### Example cfg configuration

Source: https://github.com/spotify/luigi/blob/master/doc/configuration.rst

A sample configuration file using the default cfg format, defining sections for hadoop and core settings.

```ini
[hadoop]
version=cdh4
streaming_jar=/usr/lib/hadoop-xyz/hadoop-streaming-xyz-123.jar

[core]
scheduler_host=luigi-host.mycompany.foo
```

--------------------------------

### Implement Task.run and Task.requires

Source: https://github.com/spotify/luigi/blob/master/doc/tasks.rst

Example showing a task generating a file and a downstream task consuming that file as input.

```python
class GenerateWords(luigi.Task):

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

    def run(self):

        # write a dummy list of words to output file
        words = [
                'apple',
                'banana',
                'grapefruit'
                ]

        with self.output().open('w') as f:
            for word in words:
                f.write('{word}\n'.format(word=word))


class CountLetters(luigi.Task):

    def requires(self):
        return GenerateWords()

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

    def run(self):

        # read in file as list
        with self.input().open('r') as infile:
            words = infile.read().splitlines()

        # write each word to output file with its corresponding letter count
        with self.output().open('w') as outfile:
            for word in words:
                outfile.write(
                        '{word} | {letter_count}\n'.format(
                            word=word,
                            letter_count=len(word)
                            )
                        )
```

--------------------------------

### Configure global resources in [resources]

Source: https://github.com/spotify/luigi/blob/master/doc/configuration.rst

Example configuration for defining global resource limits for the scheduler, such as hive and mysql counts.

```ini
[resources]
hive=2
mysql=1
```

--------------------------------

### Run luigid server as a daemon

Source: https://github.com/spotify/luigi/blob/master/doc/central_scheduler.rst

Command to start the Luigi daemon server with specified pidfile, log directory, and state path. Requires the python-daemon package.

```console
$ luigid --background --pidfile <PATH_TO_PIDFILE> --logdir <PATH_TO_LOGDIR> --state-path <PATH_TO_STATEFILE>
```

--------------------------------

### Trigger Luigi workflow via crontab

Source: https://github.com/spotify/luigi/blob/master/doc/execution_model.rst

Example of a cron entry to trigger the RunAll task from the my_tasks module.

```console
30 0 * * * my-user luigi RunAll --module my_tasks
```

--------------------------------

### Define a Luigi Task in Python

Source: https://github.com/spotify/luigi/blob/master/doc/running_luigi.rst

Example of a task definition in my_module.py that includes integer parameters.

```python
# my_module.py, available in your sys.path
import luigi

class MyTask(luigi.Task):
    x = luigi.IntParameter()
    y = luigi.IntParameter(default=45)

    def run(self):
        print(self.x + self.y)
```

--------------------------------

### Get Detailed Build Summary

Source: https://github.com/spotify/luigi/blob/master/doc/running_luigi.rst

Retrieve a detailed result object by setting detailed_summary=True.

```python
if __name__ == '__main__':
     luigi_run_result = luigi.build(..., detailed_summary=True)
     print(luigi_run_result.summary_text)
```

--------------------------------

### Type checking Luigi tasks with Mypy

Source: https://github.com/spotify/luigi/blob/master/doc/mypy.rst

Example of a Luigi task with typed parameters that Mypy will validate.

```python
import luigi


class MyTask(luigi.Task):
    foo: int = luigi.IntParameter()
    bar: str = luigi.Parameter()

MyTask(foo=1, bar='2')   # OK
MyTask(foo='1', bar='2')  # Error: Argument 1 to "Foo" has incompatible type "str"; expected "int"
```

--------------------------------

### Atomic write pattern - Bad approach

Source: https://github.com/spotify/luigi/blob/master/doc/luigi_patterns.rst

An example of a non-atomic write operation that causes premature completion detection.

```console
# This the BAD way
$ mkdir /outputs/final_output
$ big-slow-calculation > /outputs/final_output/foo.data
```

--------------------------------

### Backfill tasks using RangeDaily

Source: https://github.com/spotify/luigi/blob/master/doc/luigi_patterns.rst

Use the RangeDaily tool to schedule recomputation of tasks over a specific date interval by providing start and stop parameters.

```console
luigi --module all_reports RangeDaily --of AllReportsV2 --start 2014-10-31 --stop 2014-12-25
```

--------------------------------

### Define Luigi tasks for external data dependency

Source: https://github.com/spotify/luigi/blob/master/doc/execution_model.rst

Example of defining an ExternalTask for data dumps and an AggregationTask that depends on multiple past data dumps. The RunAll task serves as a dummy entry point to trigger the dependency graph.

```python
# my_tasks.py

class DataDump(luigi.ExternalTask):
    date = luigi.DateParameter()
    def output(self): return luigi.contrib.hdfs.HdfsTarget(self.date.strftime('/var/log/dump/%Y-%m-%d.txt'))

class AggregationTask(luigi.Task):
    date = luigi.DateParameter()
    window = luigi.IntParameter()
    def requires(self): return [DataDump(self.date - datetime.timedelta(i)) for i in xrange(self.window)]
    def run(self): run_some_cool_stuff(self.input())
    def output(self): return luigi.contrib.hdfs.HdfsTarget('/aggregated-%s-%d' % (self.date, self.window))

class RunAll(luigi.Task):
    ''' Dummy task that triggers execution of a other tasks'''
    def requires(self):
        for window in [3, 7, 14]:
            for d in xrange(10): # guarantee that aggregations were run for the past 10 days
               yield AggregationTask(datetime.date.today() - datetime.timedelta(d), window)
```

--------------------------------

### Define retry_count per task

Source: https://github.com/spotify/luigi/blob/master/doc/configuration.rst

Example showing how to set the retry_count field within individual Luigi task classes.

```python
class GenerateWordsFromHdfs(luigi.Task):

   retry_count = 2

    ...

class GenerateWordsFromRDBM(luigi.Task):

   retry_count = 5

    ...

class CountLetters(luigi.Task):

    def requires(self):
        return [GenerateWordsFromHdfs()]

    def run():
        yield GenerateWordsFromRDBM()

    ...
```

--------------------------------

### Build and view documentation

Source: https://github.com/spotify/luigi/blob/master/CONTRIBUTING.rst

Commands to compile the documentation using tox and open the rendered output in a browser.

```bash
tox run -e docs
```

```bash
firefox doc/_build/html/index.html
```

--------------------------------

### Run Luigi Task with Parameter Mapping

Source: https://github.com/spotify/luigi/blob/master/doc/running_luigi.rst

Demonstrates replacing underscores with hyphens for parameter names in the CLI.

```console
$ luigi --module my_module MyTask --my-parameter 100 --local-scheduler
```

--------------------------------

### Define Global Configuration Classes

Source: https://github.com/spotify/luigi/blob/master/doc/configuration.rst

Shows how to map an INI configuration section to a Python class inheriting from luigi.Config to access settings as object attributes.

```ini
[mysection]
option=hello
intoption=123
```

```python
import luigi

# Config classes should be camel cased
class mysection(luigi.Config):
    option = luigi.Parameter(default='world')
    intoption = luigi.IntParameter(default=555)

mysection().option
mysection().intoption
```

--------------------------------

### Run Top10Artists Task via CLI

Source: https://github.com/spotify/luigi/blob/master/doc/example_top_artists.rst

Execute the Top10Artists task using the Luigi command line interface with a local scheduler and specified date interval.

```console
$ luigi --module examples.top_artists Top10Artists --local-scheduler --date-interval 2012-07
```

--------------------------------

### Configure task history in luigi.cfg

Source: https://github.com/spotify/luigi/blob/master/doc/central_scheduler.rst

Configuration settings to enable task history recording in the [scheduler] section and define the database connection in the [task_history] section.

```ini
[scheduler]
record_task_history = True
state_path = /usr/local/var/luigi-state.pickle

[task_history]
db_connection = sqlite:////usr/local/var/luigi-task-hist.db
```

--------------------------------

### Configure Parameters via INI File

Source: https://github.com/spotify/luigi/blob/master/doc/parameters.rst

Set class-level parameter values using a configuration file.

```ini
[TaskA]
x: 45
```

--------------------------------

### Define MyTask with dynamic dependencies

Source: https://github.com/spotify/luigi/blob/master/doc/tasks.rst

Shows how to yield a task within the run() method to create dynamic dependencies that resolve into targets at runtime.

```python
class MyTask(luigi.Task):
    def run(self):
        other_target = yield OtherTask()

        # dynamic dependencies resolve into targets
        f = other_target.open('r')
```

--------------------------------

### Atomic write pattern - Good approach

Source: https://github.com/spotify/luigi/blob/master/doc/luigi_patterns.rst

A robust approach using a temporary directory and an atomic move operation to ensure data is only marked complete once fully written.

```console
# This is the good way
$ mkdir /outputs/final_output-tmp-123456
$ big-slow-calculation > /outputs/final_output-tmp-123456/foo.data
$ mv --no-target-directory --no-clobber /outputs/final_output{-tmp-123456,}
$ [[ -d /outputs/final_output-tmp-123456 ]] && rm -r /outputs/final_output-tmp-123456
```

--------------------------------

### Define recommended exit codes in [retcode]

Source: https://github.com/spotify/luigi/blob/master/doc/configuration.rst

Recommended configuration for Luigi binary return codes, ordered by increasing severity.

```ini
[retcode]
# The following return codes are the recommended exit codes for Luigi
# They are in increasing level of severity (for most applications)
already_running=10
missing_data=20
not_run=25
task_failed=30
scheduling_error=35
unhandled_exception=40
```

--------------------------------

### Write binary files with LocalTarget and Nop format

Source: https://github.com/spotify/luigi/blob/master/doc/tasks.rst

Use format=Nop in LocalTarget to handle binary files like pickles, as Luigi strips the 'b' flag from standard file operations.

```python
from luigi.format import Nop

class GenerateWords(luigi.Task):

    def output(self):
        return luigi.LocalTarget('words.pckl', format=Nop)

    def run(self):
        import pickle

        # write a dummy list of words to output file
        words = [
                'apple',
                'banana',
                'grapefruit'
                ]

        with self.output().open('w') as f:
            pickle.dump(words, f)
```

--------------------------------

### Initialize Visualiser Application

Source: https://github.com/spotify/luigi/blob/master/luigi/static/visualiser/index.html

JavaScript initialization call for the Luigi visualiser application pointing to the API endpoint.

```javascript
visualiserApp(new LuigiAPI("../../api"));
```

--------------------------------

### Configure Mypy via pyproject.toml

Source: https://github.com/spotify/luigi/blob/master/doc/mypy.rst

Add the luigi.mypy plugin to your pyproject.toml configuration file.

```toml
[tool.mypy]
plugins = ["luigi.mypy"]
```

--------------------------------

### Override Task Parameters via Configuration

Source: https://github.com/spotify/luigi/blob/master/doc/configuration.rst

Demonstrates defining a task with a parameter and overriding its default value using an INI configuration file.

```python
class DailyReport(luigi.contrib.hadoop.JobTask):
    date = luigi.DateParameter(default=datetime.date.today())
    # ...
```

```ini
[DailyReport]
date=2012-01-01
```

--------------------------------

### Set Class-Level Parameters

Source: https://github.com/spotify/luigi/blob/master/doc/parameters.rst

Parameters can be set at the class level via command line or configuration files.

```python
class TaskA(luigi.Task):
    x = luigi.Parameter()

class TaskB(luigi.Task):
    y = luigi.Parameter()
```

--------------------------------

### Define TaskWithManyInputs using Task.input()

Source: https://github.com/spotify/luigi/blob/master/doc/tasks.rst

Demonstrates using Task.input() to access targets from a complex structure of dependencies defined in requires().

```python
class TaskWithManyInputs(luigi.Task):
    def requires(self):
        return {'a': TaskA(), 'b': [TaskB(i) for i in xrange(100)]}

    def run(self):
        f = self.input()['a'].open('r')
        g = [y.open('r') for y in self.input()['b']]
```

--------------------------------

### Instantiate HdfsTarget

Source: https://github.com/spotify/luigi/blob/master/doc/tasks.rst

Use the HdfsTarget class to open and write to HDFS files independently of the Luigi workflow.

```python
t = luigi.contrib.hdfs.target.HdfsTarget('/tmp/test.gz', format=format.Gzip)
f = t.open('w')
# ...
f.close() # needed
```

--------------------------------

### Build Tasks from Python Code

Source: https://github.com/spotify/luigi/blob/master/doc/running_luigi.rst

Use luigi.build to execute multiple tasks programmatically.

```python
class MyTask1(luigi.Task):
    x = luigi.IntParameter()
    y = luigi.IntParameter(default=0)

    def run(self):
        print(self.x + self.y)


class MyTask2(luigi.Task):
    x = luigi.IntParameter()
    y = luigi.IntParameter(default=1)
    z = luigi.IntParameter(default=2)

    def run(self):
        print(self.x * self.y * self.z)


if __name__ == '__main__':
    luigi.build([MyTask1(x=10), MyTask2(x=15, z=3)])
```

--------------------------------

### Build Tasks with Additional Parameters

Source: https://github.com/spotify/luigi/blob/master/doc/running_luigi.rst

Pass configuration parameters like workers and local_scheduler to luigi.build.

```python
if __name__ == '__main__':
     luigi.build([MyTask1(x=1)], workers=5, local_scheduler=True)
```

--------------------------------

### Run local tests with tox

Source: https://github.com/spotify/luigi/blob/master/CONTRIBUTING.rst

Commands to execute various test suites using tox, including core tests, specific files, visualizer tests, and Hadoop-related tests.

```bash
# These commands are pretty fast and will tell if you've
# broken something major:
tox run -e py310-core

# You can also test particular files for even faster iterations
tox run -e py310-core -- test/rpc_test.py

# The visualiser tests require phantomjs to be installed on your path
tox run -e visualiser

# And some of the others involve downloading and running Hadoop:
tox run -e py310-cdh
tox run -e py310-hdp
```

--------------------------------

### Triggering multiple tasks with WrapperTask

Source: https://github.com/spotify/luigi/blob/master/doc/luigi_patterns.rst

Use WrapperTask to create a dummy task that triggers multiple dependency chains. This task does not produce output and is considered complete when all its requirements are met.

```python
class AllReports(luigi.WrapperTask):
    date = luigi.DateParameter(default=datetime.date.today())
    def requires(self):
        yield SomeReport(self.date)
        yield SomeOtherReport(self.date)
        yield CropReport(self.date)
        yield TPSReport(self.date)
        yield FooBarBazReport(self.date)
```

--------------------------------

### Run Luigi Task via CLI

Source: https://github.com/spotify/luigi/blob/master/doc/running_luigi.rst

Execute the defined task using the luigi command line tool.

```console
$ luigi --module my_module MyTask --x 123 --y 456 --local-scheduler
```

--------------------------------

### Implement a TaskContext manager for statistics reporting

Source: https://github.com/spotify/luigi/blob/master/doc/luigi_patterns.rst

Create a context manager that calls report_task_statistics in the __exit__ method to send execution metrics to the scheduler.

```python
class TaskContext:
    def __init__(self, task_process):
        self._task_process = task_process
        self._start = None

    def __enter__(self):
        self._start = time.perf_counter()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        assert self._start is not None
        elapsed = time.perf_counter() - self._start
        used_memory = max(
            resource.getrusage(resource.RUSAGE_SELF).ru_maxrss, resource.getrusage(resource.RUSAGE_CHILDREN).ru_maxrss
        )
        logging.getLogger("luigi-interface").info(
            f'Task {self._task_process.task}: time: {elapsed:.2f}s, memory: {used_memory / 1024:.2f}MB '
        )
        self._task_process.status_reporter.report_task_statistics({"memory": used_memory / 1024, "elapsed": elapsed})
```

--------------------------------

### Configure Mypy via mypy.ini

Source: https://github.com/spotify/luigi/blob/master/doc/mypy.rst

Add the luigi.mypy plugin to your mypy.ini configuration file.

```ini
[mypy]
plugins = luigi.mypy
```

--------------------------------

### Define and inspect task namespaces, families, and IDs

Source: https://github.com/spotify/luigi/blob/master/doc/tasks.rst

Demonstrates how to set a task_namespace and how it affects the task_family and task_id. Note that task_id is only available on instances, not the class itself.

```python
import luigi
class MyTask(luigi.Task):
    my_param = luigi.Parameter()
    task_namespace = 'my_namespace'

my_task = MyTask(my_param='hello')
print(my_task)                      # --> my_namespace.MyTask(my_param=hello)

print(my_task.get_task_namespace()) # --> my_namespace
print(my_task.get_task_family())    # --> my_namespace.MyTask
print(my_task.task_id)              # --> my_namespace.MyTask_hello_890907e7ce

print(MyTask.get_task_namespace())  # --> my_namespace
print(MyTask.get_task_family())     # --> my_namespace.MyTask
print(MyTask.task_id)               # --> Error!
```

--------------------------------

### Configure batch_method for parameters

Source: https://github.com/spotify/luigi/blob/master/doc/luigi_patterns.rst

Mark parameters with a batch_method in the constructor to allow the worker to combine multiple task instances into a single run.

```python
class A(luigi.Task):
    date = luigi.DateParameter(batch_method=max)
```

--------------------------------

### Run Luigi task locally

Source: https://github.com/spotify/luigi/blob/master/doc/example_top_artists.rst

Executes the AggregateArtists module locally using the local scheduler.

```console
$ cd examples
$ luigi --module top_artists AggregateArtists --local-scheduler --date-interval 2012-06
```

--------------------------------

### Task.output implementation

Source: https://github.com/spotify/luigi/blob/master/doc/tasks.rst

The output() method returns one or more Target objects. Returning a single Target is recommended to maintain atomicity.

```python
class DailyReport(luigi.Task):
    date = luigi.DateParameter()
    def output(self):
        return luigi.contrib.hdfs.HdfsTarget(self.date.strftime('/reports/%Y-%m-%d'))
    # ...
```

--------------------------------

### Run Luigi Task via Python Module

Source: https://github.com/spotify/luigi/blob/master/doc/running_luigi.rst

Execute the defined task using the python -m syntax.

```console
$ python -m luigi --module my_module MyTask --x 100 --local-scheduler
```

--------------------------------

### Build Tasks with Custom Worker Scheduler Factory

Source: https://github.com/spotify/luigi/blob/master/doc/running_luigi.rst

Provide a custom factory to control worker and scheduler implementations.

```python
class MyWorker(Worker):
    # some custom logic


class MyFactory:
  def create_local_scheduler(self):
      return scheduler.Scheduler(prune_on_get_work=True, record_task_history=False)

  def create_remote_scheduler(self, url):
      return rpc.RemoteScheduler(url)

  def create_worker(self, scheduler, worker_processes, assistant=False):
      # return your worker instance
      return MyWorker(
          scheduler=scheduler, worker_processes=worker_processes, assistant=assistant)


if __name__ == '__main__':
    luigi.build([MyTask1(x=1)], worker_scheduler_factory=MyFactory())
```

--------------------------------

### Configure Parameter Visibility

Source: https://github.com/spotify/luigi/blob/master/doc/parameters.rst

Set parameter visibility to public, hidden, or private using ParameterVisibility.

```python
>>> import luigi
>>> from luigi.parameter import ParameterVisibility

>>> luigi.Parameter(visibility=ParameterVisibility.PRIVATE)
```

--------------------------------

### Implement task message handling with accepts_messages

Source: https://github.com/spotify/luigi/blob/master/doc/luigi_patterns.rst

Configure a Luigi task to accept messages from the central scheduler by setting accepts_messages to True and checking the scheduler_messages queue in the run method.

```python
class Example(luigi.Task):

    # common task setup
    ...

    # configure the task to accept all incoming messages
    accepts_messages = True

    def run(self):
        # this example runs some loop and listens for the
        # "terminate" message, and responds to all other messages
        for _ in some_loop():
            # check incomming messages
            if not self.scheduler_messages.empty():
                msg = self.scheduler_messages.get()
                if msg.content == "terminate":
                    break
                else:
                    msg.respond("unknown message")

        # finalize
        ...
```

--------------------------------

### Traceback Modal Template

Source: https://github.com/spotify/luigi/blob/master/luigi/static/visualiser/index.html

Modal template for displaying task error tracebacks and the corresponding re-run command.

```html
<div class="modal-dialog"> <div class="modal-content"> <div class="modal-header"> <button type="button" class="close" data-dismiss="modal"><span aria-hidden="true">&times;</span><span class="sr-only">Close</span></button> <h4 class="modal-title" id="myModalLabel">Traceback for {{displayName}}</h4> </div> <div class="modal-body"> <pre class="pre-scrollable">{{error}}</pre> Command to re-run: <pre class="pre-scrollable">luigi --module {{taskModule}} {{taskFamily}} {{taskParams}}</pre> </div> <div class="modal-footer"> <button type="button" class="btn btn-default" data-dismiss="modal">Close</button> </div> </div> </div>
```

--------------------------------

### Render Resource Box and Task Table

Source: https://github.com/spotify/luigi/blob/master/luigi/static/visualiser/index.html

Mustache template for rendering resource progress bars and associated task tables with action links.

```html
{{#num_consumer}} <i class="fa fa-navicon resources-collapse" data-target="#collapse-{{name}}"></i> {{/num_consumer}} </div><!-- /.box-tools --> </div><!-- /.box-header --> <div class="box-body" id="{{name}}-resource-box"> <div class="progress"> <div class="progress-bar progress-bar-{{bar_type}}" style="width: {{percent_used}}%"> <b>{{num_used}}/{{num_total}}</b> </div> </div> {{#num_consumer}} <div class="collapse resource-box" id="collapse-{{name}}" data-resource="{{name}}"> <table class="table table-striped worker-table"> <thead> <th>Name</th> <th>Priority</th> <th>Time</th> <th>Actions</th> </thead> <tbody> {{#tasks}} <tr> <td>{{displayName}}</td> <td>{{priority}}</td> <td>{{displayTime}}</td> <td><a href="#tab=graph&taskId={{taskId}}&hideDone=1" class="btn btn-info btn-xs" title="View graph" data-toggle="tooltip" data-action="drawGraph"><i class="fa fa-sitemap"/></a></td> </tr> {{/tasks}} </tbody> </table> </div> {{/num_consumer}} </div><!-- /.box-body --> </div><!-- /.box --> {{/resources}}
```

--------------------------------

### Define ArtistToplistToDatabase Task

Source: https://github.com/spotify/luigi/blob/master/doc/example_top_artists.rst

A subclass of luigi.contrib.postgres.CopyToTable that defines database connection parameters and columns, while requiring the Top10Artists task.

```python
class ArtistToplistToDatabase(luigi.contrib.postgres.CopyToTable):
    date_interval = luigi.DateIntervalParameter()
    use_hadoop = luigi.BoolParameter()

    host = "localhost"
    database = "toplists"
    user = "luigi"
    password = "abc123"  # ;)
    table = "top10"

    columns = [("date_from", "DATE"),
               ("date_to", "DATE"),
               ("artist", "TEXT"),
               ("streams", "INT")]

    def requires(self):
        return Top10Artists(self.date_interval, self.use_hadoop)
```

--------------------------------

### Run Luigi task with PYTHONPATH

Source: https://github.com/spotify/luigi/blob/master/doc/example_top_artists.rst

Executes the task while explicitly setting the PYTHONPATH to the current directory to avoid ImportError.

```console
$ PYTHONPATH='.' luigi --module top_artists AggregateArtists --local-scheduler --date-interval 2012-06
```

--------------------------------

### AggregateArtists Task Implementation

Source: https://github.com/spotify/luigi/blob/master/doc/example_top_artists.rst

A Luigi task that aggregates artist streams from input files and writes the counts to a local TSV file. It requires a date interval parameter and depends on a Streams task.

```python
class AggregateArtists(luigi.Task):
    date_interval = luigi.DateIntervalParameter()

    def output(self):
        return luigi.LocalTarget("data/artist_streams_%s.tsv" % self.date_interval)

    def requires(self):
        return [Streams(date) for date in self.date_interval]

    def run(self):
        artist_count = defaultdict(int)

        for input in self.input():
            with input.open('r') as in_file:
                for line in in_file:
                    timestamp, artist, track = line.strip().split()
                    artist_count[artist] += 1

        with self.output().open('w') as out_file:
            for artist, count in artist_count.iteritems():
                print(artist, count, file=out_file)
```

--------------------------------

### AggregateArtistsSpark task implementation

Source: https://github.com/spotify/luigi/blob/master/doc/example_top_artists.rst

Defines a Luigi task that orchestrates a Spark job to aggregate artist data from HDFS.

```python
class AggregateArtistsSpark(luigi.contrib.spark.SparkSubmitTask):
    date_interval = luigi.DateIntervalParameter()

    app = 'top_artists_spark.py'
    master = 'local[*]'

    def output(self):
        return luigi.contrib.hdfs.HdfsTarget("data/artist_streams_%s.tsv" % self.date_interval)

    def requires(self):
        return [StreamsHdfs(date) for date in self.date_interval]

    def app_options(self):
        # :func:`~luigi.task.Task.input` returns the targets produced by the tasks in
        # `~luigi.task.Task.requires`.
        return [','.join([p.path for p in self.input()]),
                self.output().path]
```

--------------------------------

### Render Navigation Link

Source: https://github.com/spotify/luigi/blob/master/luigi/static/visualiser/index.html

Mustache template for rendering individual navigation links in the sidebar.

```html
<li> <a class="js-nav-link" href="{{href}}" {{#dataTab}}data-tab="{{dataTab}}"{{/dataTab}}> {{label}} </a> </li>
```

--------------------------------

### Set task status and progress in Task.run

Source: https://github.com/spotify/luigi/blob/master/doc/tasks.rst

Use set_tracking_url, set_status_message, and set_progress_percentage within the run method to update the central scheduler GUI.

```python
class MyTask(luigi.Task):
    def run(self):
        # set a tracking url
        self.set_tracking_url("http://...")

        # set status messages during the workload
        for i in range(100):
            # do some hard work here
            if i % 10 == 0:
                self.set_status_message("Progress: %d / 100" % i)
                # displays a progress bar in the scheduler UI
                self.set_progress_percentage(i)
```

--------------------------------

### Task Action Buttons Template

Source: https://github.com/spotify/luigi/blob/master/luigi/static/visualiser/index.html

Template for rendering task-specific action buttons like graph viewing, error inspection, and task re-enabling.

```html
<div class="span2"> <a href="#tab=graph&taskId={{encodedTaskId}}&hideDone=1" class="btn btn-info btn-xs" title="View graph" data-toggle="tooltip"><i class="fa fa-sitemap"></i></a> {{#error}}<button class="btn btn-danger btn-xs showError" title="Show error" data-toggle="tooltip"><i class="fa fa-bug"></i></button>{{/error}} {{#error}}<button class="btn btn-warning btn-xs forgiveFailures" title="Forgive failures" data-toggle="tooltip"><i class="fa fa-ambulance"></i></button>{{/error}} {{#re\_enable}}<button class="btn btn-danger btn-xs showError" title="Show error" data-toggle="tooltip"><i class="fa fa-bug"></i></button>{{/re\_enable}} {{#re\_enable}}<a class="btn btn-warning btn-xs re-enable-button" title="Re-enable" data-toggle="tooltip" data-task-id="{{taskId}}">Re-enable</a>{{/re\_enable}} {{#mark\_as\_done}}<a class="btn btn-success btn-xs markAsDone" title="Mark as done" data-toggle="tooltip" data-task-id="{{taskId}}"><i class="fa fa-fast-forward"></i></a>{{/mark\_as\_done}} {{#trackingUrl}}<a target="\_blank" href="{{trackingUrl}}" class="btn btn-primary btn-xs" title="Track Progress" data-toggle="tooltip"><i class="fa fa-eye"></i></a>{{/trackingUrl}} {{#statusMessage}}<button class="btn btn-primary btn-xs statusMessage" title="Status message" data-toggle="tooltip" data-task-id="{{taskId}}" data-display-name="{{displayName}}"><i class="fa fa-comment"></i></button>{{/statusMessage}} {{^statusMessage}} {{#progressPercentage}}<button class="btn btn-primary btn-xs statusMessage" title="Status message" data-toggle="tooltip" data-task-id="{{taskId}}" data-display-name="{{displayName}}"><i class="fa fa-comment"></i></button> {{/progressPercentage}} {{/statusMessage}} {{#acceptsMessages}}<button class="btn btn-default btn-xs schedulerMessage" title="Send message" data-toggle="tooltip" data-task-id="{{taskId}}" data-display-name="{{displayName}}" data-worker="{{workerIdRunning}}"><i class="fa fa-paper-plane"></i></button>{{/acceptsMessages}} </div>
```

--------------------------------

### ExternalTask with parameters

Source: https://github.com/spotify/luigi/blob/master/doc/tasks.rst

Add parameters to an ExternalTask to dynamically define the Target path.

```python
class LogFiles(luigi.ExternalTask):
    date = luigi.DateParameter()
    def output(self):
        return luigi.contrib.hdfs.HdfsTarget(self.date.strftime('/log/%Y-%m-%d'))
```

--------------------------------

### Define a Task Parameter

Source: https://github.com/spotify/luigi/blob/master/doc/parameters.rst

Declare parameters by instantiating Parameter objects on the class scope to enable automatic constructor and command-line handling.

```python
class DailyReport(luigi.contrib.hadoop.JobTask):
    date = luigi.DateParameter(default=datetime.date.today())
    # ...
```

--------------------------------

### Access Parameter Values

Source: https://github.com/spotify/luigi/blob/master/doc/parameters.rst

Parameter values are set on the task instance upon construction or command-line invocation.

```python
d = DailyReport(datetime.date(2012, 5, 10))
print(d.date)
```

--------------------------------

### Create a custom MetricsCollector for task statistics

Source: https://github.com/spotify/luigi/blob/master/doc/luigi_patterns.rst

Implement a custom metrics collector by inheriting from MetricsCollector and overriding handle_task_statistics to process task execution data.

```python
class MetricsCollector(PrometheusMetricsCollector):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.task_run_execution_time = Gauge(
            'luigi_task_run_execution_time_seconds',
            'luigi task run method execution time in seconds',
            self.labels,
            registry=self.registry
        )
        self.task_execution_memory = Gauge(
            'luigi_task_max_memory_megabytes',
            'luigi task run method max memory usage in megabytes',
            self.labels,
            registry=self.registry
        )

    def handle_task_statistics(self, task, statistics):
        if "elapsed" in statistics:
            self.task_run_execution_time.labels(**self._generate_task_labels(task)).set(statistics["elapsed"])
        if "memory" in statistics:
            self.task_execution_memory.labels(**self._generate_task_labels(task)).set(statistics["memory"])
```

--------------------------------

### Define Top10Artists Luigi Task

Source: https://github.com/spotify/luigi/blob/master/doc/example_top_artists.rst

This task defines a dependency on either AggregateArtistsSpark or AggregateArtists based on the use_hadoop parameter, then processes the input to find the top 10 artists.

```python
class Top10Artists(luigi.Task):
    date_interval = luigi.DateIntervalParameter()
    use_hadoop = luigi.BoolParameter()

    def requires(self):
        if self.use_hadoop:
            return AggregateArtistsSpark(self.date_interval)
        else:
            return AggregateArtists(self.date_interval)

    def output(self):
        return luigi.LocalTarget("data/top_artists_%s.tsv" % self.date_interval)

    def run(self):
        top_10 = nlargest(10, self._input_iterator())
        with self.output().open('w') as out_file:
            for streams, artist in top_10:
                print(self.date_interval.date_a, self.date_interval.date_b, artist, streams, file=out_file)

    def _input_iterator(self):
        with self.input().open('r') as in_file:
            for line in in_file:
                artist, streams = line.strip().split()
                yield int(streams), int(artist)
```

--------------------------------

### Render Task Families Sidebar

Source: https://github.com/spotify/luigi/blob/master/luigi/static/visualiser/index.html

Mustache template for generating the sidebar menu of task families and their counts.

```html
<ul class="sidebar-menu"> <li class="header">TASK FAMILIES <span class="badge" id="clear-task-filter">Clear selection</span></li> {{#tasks}} <li> <a href="#" class="sidebar-folder">{{name}}</a> <ul class="sidebar-menu" style="display: none"> {{#tasks}} <li data-task="{{name}}"><a href="#"><span class="badge">{{count}}</span> {{name}}</a> </li> {{/tasks}} </ul> </li> {{/tasks}} </ul>
```

--------------------------------

### Sticky Footer and Layout CSS

Source: https://github.com/spotify/luigi/blob/master/luigi/templates/layout.html

Defines the layout structure including a sticky footer, responsive adjustments, and container constraints.

```css
/* Sticky footer styles -------------------------------------------------- */ html, body { height: 100%; /* The html and body elements cannot have any padding or margin. */ } /* Wrapper for page content to push down footer */ #wrap { min-height: 90%; /*height: auto !important;*/ height: 730px; /* Negative indent footer by it's height */ margin: 0 auto -60px; overflow: scroll; } /* Set the fixed height of the footer here */ #push, #footer { height: 60px; } #footer { background-color: #f5f5f5; } /* Lastly, apply responsive CSS fixes as necessary */ @media (max-width: 767px) { #footer { margin-left: -20px; margin-right: -20px; padding-left: 20px; padding-right: 20px; } } /* Custom page CSS -------------------------------------------------- */ /* Not required for template or sticky footer method. */ .container { width: auto; max-width: 1200px; margin-bottom: 20px; } .container .credit { margin: 20px 0; } .container[role="main"] { padding-bottom: 60px; } #footer { position: fixed; bottom: 0; left: 0; right: 0; } #proper-content { overflow: auto; } .lead { margin-top: -17px; margin-bottom: 13px; }
```

--------------------------------

### Render Task Display Callouts

Source: https://github.com/spotify/luigi/blob/master/luigi/static/visualiser/index.html

Mustache templates for displaying status callouts regarding missing categories or active filters.

```html
<div class="callout callout-danger"> <p> Too many {{#missingCategories}} <strong>{{name}}</strong>, {{/missingCategories}} tasks to display. </p> <p> Task family counts only include displayed tasks. Use <strong>Filter on Server</strong> to constrain search. </p> </div> <div class="callout callout-info"> Displaying {{#catNames}}<strong>{{ name }}</strong>, {{/catNames}} tasks {{#taskFamily}}of family <strong>{{taskFamily}}</strong>{{/taskFamily}} {{#tableFilter}}filtered by <strong>"{{tableFilter}}"</strong>{{/tableFilter}}. </div>
```

--------------------------------

### Dynamic resource locking with Python properties

Source: https://github.com/spotify/luigi/blob/master/doc/luigi_patterns.rst

Define resources as a Python property to dynamically lock specific files based on task parameters, preventing simultaneous writes to the same file.

```python
class A(luigi.Task):
    ...

    @property
    def resources(self):
        return { self.important_file_name: 1 }
```

--------------------------------

### Configure task priority

Source: https://github.com/spotify/luigi/blob/master/doc/tasks.rst

Set task execution order by defining a static priority class constant or a dynamic priority property. Higher values are prioritized by the scheduler, with a default of 0.

```python
# A static priority value as a class constant:
class MyTask(luigi.Task):
    priority = 100
    # ...

# A dynamic priority value with a "@property" decorated method:
class OtherTask(luigi.Task):
    @property
    def priority(self):
        if self.date > some_threshold:
            return 80
        else:
            return 40
    # ...
```

--------------------------------

### Propagate parameters with DictParameter

Source: https://github.com/spotify/luigi/blob/master/doc/luigi_patterns.rst

Pass additional parameters to tasks within a Range class using the of-params flag with a JSON-encoded dictionary.

```console
luigi RangeDaily --of MyTask --start 2014-10-31 --of-params '{"my_string_param": "123", "my_int_param": 123}'
```

--------------------------------

### Jinja2 Content Block

Source: https://github.com/spotify/luigi/blob/master/luigi/templates/layout.html

Defines the content block placeholder for the template.

```html
{% block content %}{% end %}
```

--------------------------------

### Task.requires implementation

Source: https://github.com/spotify/luigi/blob/master/doc/tasks.rst

The requires() method specifies dependencies on other Task objects, which can be returned in various structures like lists or tuples.

```python
def requires(self):
    return OtherTask(self.date), DailyReport(self.date - datetime.timedelta(1))
```

--------------------------------

### Status Message Modal Template

Source: https://github.com/spotify/luigi/blob/master/luigi/static/visualiser/index.html

Modal template for displaying task status messages and progress bars.

```html
<div class="modal-dialog"> <div class="modal-content"> <div class="modal-header"> <button type="button" class="close" data-dismiss="modal"><span aria-hidden="true">&times;</span><span class="sr-only">Close</span></button> <h4 class="modal-title" id="myModalLabel">Status message for {{displayName}}</h4> </div> <div class="modal-body"> <pre class="pre-scrollable">{{statusMessage}}</pre> <div class="progress"> <div class="progress-bar" role="progressbar" aria-valuenow="{{progressPercentage}}" aria-valuemin="0" aria-valuemax="100" style="min-width: 2em;"> {{progressPercentage}}% </div> </div> </div> <div class="modal-footer"> <button type="button" class="btn btn-default" data-dismiss="modal">Close</button> </div> </div> </div>
```

--------------------------------

### ExternalTask wrapper for Target objects

Source: https://github.com/spotify/luigi/blob/master/doc/tasks.rst

Wrap an externally created Target object in an ExternalTask class to use it as a dependency.

```python
class LogFiles(luigi.ExternalTask):
    def output(self):
        return luigi.contrib.hdfs.HdfsTarget('/log')
```

--------------------------------

### Aggregate multiple batch parameters

Source: https://github.com/spotify/luigi/blob/master/doc/luigi_patterns.rst

When multiple parameters have batch methods, they are aggregated separately based on their respective methods.

```python
class A(luigi.Task):
    p1 = luigi.IntParameter(batch_method=max)
    p2 = luigi.IntParameter(batch_method=max)
    p3 = luigi.IntParameter()
```

--------------------------------

### Run JobTask manually

Source: https://github.com/spotify/luigi/blob/master/doc/tasks.rst

Execute a Hadoop JobTask directly without the full Luigi workflow scheduler.

```python
MyJobTask('abc', 123).run()
```

--------------------------------

### Set Workers Modal Template

Source: https://github.com/spotify/luigi/blob/master/luigi/static/visualiser/index.html

Modal template for configuring the number of workers.

```html
<div class="modal fade" id="setWorkersModal" tabindex="-1" role="dialog" aria-labelledby="setWorkersLabel"> <div class="modal-dialog" role="document"> <div class="modal-content"> <div class="modal-header"> <button type="button" class="close" data-dismiss="modal" aria-label="Close"><span aria-hidden="true">&times;</span></button> <h4 class="modal-title" id="setWorkersLabel">Set workers</h4> </div> <div class="modal-body"> <form>
```

--------------------------------

### Send Message Modal Template

Source: https://github.com/spotify/luigi/blob/master/luigi/static/visualiser/index.html

Modal template for sending messages to tasks, including input fields and response display.

```html
<div class="modal-dialog"> <div class="modal-content"> <div class="modal-header"> <button type="button" class="close" data-dismiss="modal"><span aria-hidden="true">&times;</span><span class="sr-only">Close</span></button> <h4 class="modal-title" id="myModalLabel">Send message to {{displayName}}</h4> </div> <div class="modal-body"> <form> <div class="form-group"> <label for="schedulerMessageInput">Message:</label> <input type="text" class="form-control" id="schedulerMessageInput" placeholder=""> </div> <div class="form-group"> <input type="checkbox" class="form-check-input" id="schedulerMessageAwaitResponse"> <label class="form-check-label" for="schedulerMessageAwaitResponse">Await response</label> </div> </form> <div class="form-group" id="schedulerMessageResponse" style="display: none;"> <hr /> <label>Response:</label> <pre class="pre-scrollable"><i class="fa fa-spinner fa-pulse" id="schedulerMessageSpinner"></i><div></div></pre> </div> </div> <div class="modal-footer"> <button type="button" class="btn btn-default" data-dismiss="modal">Cancel</button> <button type="button" id="schedulerMessageButton" data-dismiss="modal" class="btn btn-primary">Send</button> </div> </div> </div>
```

--------------------------------

### Spark job processing code

Source: https://github.com/spotify/luigi/blob/master/doc/example_top_artists.rst

Python script for the Spark job that reads input streams and calculates artist counts.

```python
import operator
import sys
from pyspark.sql import SparkSession


def main(argv):
    input_paths = argv[1].split(',')
    output_path = argv[2]

    spark = SparkSession.builder.getOrCreate()

    streams = spark.read.option('sep', '\t').csv(input_paths[0])
    for stream_path in input_paths[1:]:
        streams.union(spark.read.option('sep', '\t').csv(stream_path))

    # The second field is the artist
    counts = streams \
        .map(lambda row: (row[1], 1)) \
        .reduceByKey(operator.add)

    counts.write.option('sep', '\t').csv(output_path)


if __name__ == '__main__':
    sys.exit(main(sys.argv))
```

--------------------------------

### Trigger recurring tasks with RangeDaily

Source: https://github.com/spotify/luigi/blob/master/doc/luigi_patterns.rst

Use RangeDaily for efficient scheduling of hundreds of tasks over long periods. It requires the task to implement bulk_complete or use file system Targets with date-parameterized paths.

```console
luigi --module all_reports RangeDaily --of AllReports --start 2015-01-01
```

--------------------------------

### Task Instance Caching

Source: https://github.com/spotify/luigi/blob/master/doc/parameters.rst

Luigi caches task instances; two tasks of the same class with identical parameter values are the same instance.

```python
>>> import luigi
>>> import datetime
>>> class DateTask(luigi.Task):
...   date = luigi.DateParameter()
...
>>> a = datetime.date(2014, 1, 21)
>>> b = datetime.date(2014, 1, 21)
>>> a is b
False
>>> c = DateTask(date=a)
>>> d = DateTask(date=b)
>>> c
DateTask(date=2014-01-21)
>>> d
DateTask(date=2014-01-21)
>>> c is d
True
```

--------------------------------

### Register event handlers for task lifecycle

Source: https://github.com/spotify/luigi/blob/master/doc/tasks.rst

Use the event_handler decorator to register callbacks for specific events like SUCCESS or FAILURE on Task classes.

```python
@luigi.Task.event_handler(luigi.Event.SUCCESS)
def celebrate_success(task):
    """Will be called directly after a successful execution
       of `run` on any Task subclass (i.e. all luigi Tasks)
    """
    ...

@luigi.contrib.hadoop.JobTask.event_handler(luigi.Event.FAILURE)
def mourn_failure(task, exception):
    """Will be called directly after a failed execution
       of `run` on any JobTask subclass
    """
    ...

luigi.run()
```

--------------------------------

### Propagate parameters at task family level

Source: https://github.com/spotify/luigi/blob/master/doc/luigi_patterns.rst

Specify parameters directly at the task family level, though note that these will not appear in the task name for the upstream Range task.

```console
luigi RangeDaily --of MyTask --start 2014-10-31 --MyTask-my-param 123
```

--------------------------------

### Limit batch size with max_batch_size

Source: https://github.com/spotify/luigi/blob/master/doc/luigi_patterns.rst

Set the max_batch_size attribute to limit the number of jobs combined in a single batch.

```python
class A(luigi.Task):
    date = luigi.DateParameter(batch_method=max)

    max_batch_size = 10
```

--------------------------------

### Use Insignificant Parameters

Source: https://github.com/spotify/luigi/blob/master/doc/parameters.rst

Parameters marked with significant=False are ignored in the task signature, meaning tasks with different values for these parameters are not the same instance but share the same signature.

```python
>>> class DateTask2(DateTask):
...   other = luigi.Parameter(significant=False)
...
>>> c = DateTask2(date=a, other="foo")
>>> d = DateTask2(date=b, other="bar")
>>> c
DateTask2(date=2014-01-21)
>>> d
DateTask2(date=2014-01-21)
>>> c.other
'foo'
>>> d.other
'bar'
>>> c is d
False
>>> hash(c) == hash(d)
True
```

--------------------------------

### Decrease task resources dynamically

Source: https://github.com/spotify/luigi/blob/master/doc/luigi_patterns.rst

Use self.decrease_running_resources within the run method to release resources after heavy computation, allowing other tasks to proceed.

```python
class A(luigi.Task):

    # set maximum resources a priori
    resources = {"some_resource": 3}

    def run(self):
        # do something
        ...

        # decrease consumption of "some_resource" by one
        self.decrease_running_resources({"some_resource": 1})

        # continue with reduced resources
        ...
```

--------------------------------

### Preventing concurrent overwrites with batch_method and resources

Source: https://github.com/spotify/luigi/blob/master/doc/luigi_patterns.rst

Use batch_method set to max and a unique resource to ensure only the latest task runs when overwriting a data source.

```python
class A(luigi.Task):
    date = luigi.DateParameter(batch_method=max)

    resources = {'overwrite_resource': 1}
```

--------------------------------

### Trigger recurring tasks with RangeDailyBase

Source: https://github.com/spotify/luigi/blob/master/doc/luigi_patterns.rst

Use RangeDailyBase in a crontab to ensure tasks are completed for every date, preventing gaps in deliverables. By default, it checks up to 3 months of history.

```console
luigi --module all_reports RangeDailyBase --of AllReports --start 2015-01-01
```
Tokens:
Raw