Implementation:Spotify Luigi ExternalTask LocalTarget

Overview

Concrete tool for declaring pre-existing data dependencies that already exist on the local filesystem, provided by Luigi.

Description

Luigi provides two classes that work together to represent external data on the local filesystem:

ExternalTask -- A subclass of Task whose run attribute is set to None. This signals to the Luigi framework that the task produces no computation; it merely asserts that certain data exists. When the worker encounters an ExternalTask, it checks completeness via output().exists() rather than attempting to execute a run() method.

LocalTarget -- A concrete FileSystemTarget that wraps a path on the local filesystem. It provides atomic write operations (via a temporary file that is moved into place on close), format-aware reading and writing, and an exists() method that delegates to os.path.exists().

Together, a subclass of ExternalTask whose output() returns a LocalTarget represents the canonical way to declare "this local file should already exist."

Usage

Use ExternalTask with LocalTarget when:

Your pipeline depends on a file that is produced by an external process (e.g., a daily CSV drop).
You want Luigi's scheduler to verify the file's existence before running downstream tasks.
You need atomic read/write semantics for local files within your pipeline.

Code Reference

ExternalTask

Attribute	Value
Source Location	`luigi/task.py`, lines 861-869
Signature	`class ExternalTask(Task)`
Import	`from luigi import ExternalTask` or `import luigi; luigi.ExternalTask`

ExternalTask inherits from Task and overrides a single attribute:

class ExternalTask(Task):
    """
    Subclass for references to external dependencies.

    An ExternalTask's does not have a `run` implementation, which signifies to
    the framework that this Task's output is generated outside of Luigi.
    """
    run = None

LocalTarget

Attribute	Value
Source Location	`luigi/local_target.py`, lines 131-191
Signature	`class LocalTarget(FileSystemTarget)`
Constructor	`def __init__(self, path=None, format=None, is_tmp=False)`
Import	`from luigi import LocalTarget` or `import luigi; luigi.LocalTarget`

class LocalTarget(FileSystemTarget):
    fs = LocalFileSystem()

    def __init__(self, path=None, format=None, is_tmp=False):
        if format is None:
            format = get_default_format()
        if not path:
            if not is_tmp:
                raise Exception('path or is_tmp must be set')
            path = os.path.join(
                tempfile.gettempdir(),
                'luigi-tmp-%09d' % random.randint(0, 999999999),
            )
        super(LocalTarget, self).__init__(path)
        self.format = format
        self.is_tmp = is_tmp

Key methods:

open(mode='r') -- Opens the target for reading ('r') or writing ('w'). Write mode uses an atomic temporary file.
exists() -- Returns True if os.path.exists(self.path).
makedirs() -- Creates all parent directories of self.path.

I/O Contract

ExternalTask

Direction	Name	Type	Description
Input	(none)	--	ExternalTask accepts no inputs; it is a leaf node.
Output	return value of `output()`	`Target` (e.g. `LocalTarget`)	The external data asset to check for existence.

LocalTarget

Direction	Name	Type	Description
Input	`path`	`str` or `None`	Filesystem path to the target file. Required unless `is_tmp=True`.
Input	`format`	`Format` or `None`	File format for pipe-based reading/writing. Defaults to `get_default_format()`.
Input	`is_tmp`	`bool`	If `True`, creates a temporary path and auto-deletes on garbage collection.
Output	file stream	file-like object	Returned by `open('r')` or `open('w')`.

Usage Examples

Declaring an external file dependency

This example is modelled on the InputText class from examples/wordcount.py in the Luigi repository.

import luigi


class InputText(luigi.ExternalTask):
    """
    Represents a text file that is produced outside of this pipeline.
    The file is expected to exist at /var/tmp/text/YYYY-MM-DD.txt.
    """
    date = luigi.DateParameter()

    def output(self):
        return luigi.LocalTarget(
            self.date.strftime('/var/tmp/text/%Y-%m-%d.txt')
        )


class WordCount(luigi.Task):
    date_interval = luigi.DateIntervalParameter()

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

    def output(self):
        return luigi.LocalTarget(
            '/var/tmp/text-count/%s' % self.date_interval
        )

    def run(self):
        count = {}
        for f in self.input():
            for line in f.open('r'):
                for word in line.strip().split():
                    count[word] = count.get(word, 0) + 1

        with self.output().open('w') as out:
            for word, cnt in count.items():
                out.write("%s\t%d\n" % (word, cnt))

Atomic writes with LocalTarget

import luigi


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

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

    def run(self):
        # open('w') writes to a temporary file first, then atomically
        # moves it to the final path when close() or __exit__ is called.
        with self.output().open('w') as f:
            f.write("metric,value\n")
            f.write("users,1234\n")
            f.write("sessions,5678\n")

Related Pages

Page Connections

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

Principle

Implementation

Heuristic

Environment