DOCSEARCHbyAlgolia
Submit Docs

Luigi

repository·master·Indexed 12 days ago

https://github.com/spotify/luigi

A Python-based framework for building complex dependency graphs of tasks, designed for workflow management, task scheduling, and dependency resolution in data pipelines.

Tokens
47.8K
Snippets
150
Records
216
Agent score
97%

What's inside Luigi

  1. What is Luigi?

    master

    Luigi is a Python package designed to help build complex pipelines of batch jobs. It manages the 'plumbing' of long-running batch processes, including:

    • Dependency Resolution: Automatically determining the order in which tasks must run.
    • Workflow Management: Orchestrating the execution of tasks.
    • Visualization: Providing a web-based interface to view the dependency graph.
    • Failure Handling: Managing what happens when tasks fail.
    • Command Line Integration: Running tasks via the CLI.

    Luigi is not a data processing framework like Hive, Pig, or Spark; instead, it acts as the glue that stitches various tasks (such as Hive queries, Spark jobs, or Python snippets) together into a cohesive pipeline.

  2. Understand parameter resolution order

    master

    When a task is instantiated, Luigi resolves parameter values using the following priority (from highest to lowest):

    1. Constructor/Instance level: Values passed directly to the task constructor in Python, or task-level values set on the command line.
    2. Command Line (Class level): Values set via the CLI using the --TaskName-parameter syntax.
    3. Configuration File: Values defined in the configuration file under the [TaskName] section.
    4. Default Value: The default value provided when the parameter was defined in the class.

  3. Define a basic Luigi Task

    master

    To create a task in Luigi, subclass luigi.Task and implement the following methods:

    • output(): Returns a Target object (e.g., luigi.LocalTarget) representing where the task's results are stored. This allows Luigi to check if the task has already successfully completed.
    • requires(): Returns a list of tasks or targets that must be completed before this task can run.
    • run(): Contains the actual logic of the task. This method is called if the output() target does not exist.

    Tasks are idempotent: running them multiple times results in the same outcome. If the output file exists, Luigi will skip the task unless the file is manually deleted.

    import luigi

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):
        # Task logic goes here
        with self.output().open('w') as out_file:
            out_file.write("some data")

  4. How Task instance caching works with parameters

    master

    Luigi identifies tasks uniquely by their class name and the values of their parameters. Within the same worker, two tasks of the same class with identical parameter values are treated as the exact same instance (c is d will be True).

    Insignificant parameters

    If you define a parameter with significant=False, it is ignored when determining the task's identity (signature). Tasks that differ only by insignificant parameters will have the same hash and signature, but they remain distinct objects in memory.

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

class DateTask2(DateTask):
    other = luigi.Parameter(significant=False)

# c and d have the same signature/hash but are different instances
c = DateTask2(date=datetime.date(2014, 1, 21), other="foo")
d = DateTask2(date=datetime.date(2014, 1, 21), other="bar")

print(c is d)      # False
print(hash(c) == hash(d)) # True

  5. Understand logging configuration resolution order

    master

    If multiple logging configurations are provided, Luigi resolves them in the following order of precedence (highest priority first):

    1. no_configure_logging option
    2. --background (Luigid CLI)
    3. --logdir (Luigid CLI)
    4. --logging-conf-file (Worker CLI)
    5. logging_conf_file option (in [core] config)
    6. [logging] section (in TOML config)
    7. --log-level (Worker CLI)
    8. log_level option (in [core] config)

  6. Batch multiple parameter values into a single run

    master

    If running multiple jobs together is more efficient than running them individually, use the batch_method in a parameter's constructor. This tells the scheduler how to combine multiple task instances into one.

    Using batch_method=max

    Commonly used for tasks that overwrite older data. If multiple tasks with different dates are ready, the scheduler will run only the one with the maximum value and mark the others as batch_running.

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

    Controlling Batch Size

    You can limit the maximum number of tasks in a batch using max_batch_size:

    class A(luigi.Task):
    date = luigi.DateParameter(batch_method=max)
    max_batch_size = 10

    Preventing Concurrent Writes

    If tasks in a batch overwrite the same data source, use a unique resource to ensure only one batch runs at a time:

    class A(luigi.Task):
    date = luigi.DateParameter(batch_method=max)
    resources = {'overwrite_resource': 1}

    Avoiding concurrent writes to a single file

    If multiple tasks must update the same file, turn resources into a property that returns a value based on the file name. Since the default limit is 1, no two tasks with the same file name will run simultaneously.

    class A(luigi.Task):
    @property
    def resources(self):
        return { self.important_file_name: 1 }

  7. How Luigi works: The Python-based dependency model

    master

    Luigi's core philosophy is that the dependency graph is specified within Python code rather than in external XML or configuration files.

    This approach allows you to leverage Python's full power to define complex dependencies, such as:

    • Date Algebra: Dynamically calculating dependencies based on time.
    • Recursive References: Referencing other versions of the same task.
    • Extensibility: While the orchestration is in Python, Luigi can trigger non-Python processes like Pig scripts, SSH commands, or Hadoop jobs.

    Conceptually, it functions similarly to GNU Make, where tasks are defined and their dependencies are resolved before execution.

  8. Trigger multiple tasks using WrapperTask

    master

    If you want to trigger a large group of pipelines by specifying a single task in the command line (similar to make), use luigi.WrapperTask. Unlike a standard luigi.Task, a WrapperTask does not produce its own output; it is considered complete once all the tasks it requires() are finished.

    This is useful for creating 'entry point' tasks that aggregate various dependency chains.

    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)

  9. Ensure atomic writes to prevent partial data corruption

    master

    A common mistake in Luigi is writing data partially to a final destination. Because Luigi's completion checks (like Task.complete) often rely on whether a target exists (e.g., a folder), a task might be marked as complete while its data is still being written. This can cause downstream tasks to consume incomplete or corrupt data.

    To avoid this, you must ensure your writes are atomic. If you are using a file system, use luigi.target.FileSystemTarget.temporary_path to write to a temporary location first and then move it to the final destination in a single atomic operation.

    For non-file system targets (databases, HDFS, etc.), you must implement your own atomic write logic (e.g., writing to a temporary table and then renaming/swapping).

    # If using FileSystemTarget, use the built-in temporary path helper
target = FileSystemTarget('/outputs/final_output/foo.data')
with target.temporary_path() as tmp_path:
    # Perform the slow calculation writing to tmp_path
    run_big_calculation(tmp_path)
# Once the block exits, Luigi handles the atomic move to the final target

  10. Understand the Luigi execution model

    master

    Luigi uses a model where the worker that schedules the tasks also executes them within the same process. This means no execution is transferred by default (unless a specific task type like HadoopJobTask is explicitly designed to do so).

    Key Characteristics:

    • Debugging: Since execution happens within the worker process, debugging is straightforward.
    • Deployment: Deployment is simplified because the same process used in development can be used in production.
    • Scalability: Luigi does not provide automatic scalability for free. It is highly efficient for managing dependency graphs, but scaling to thousands of concurrent tasks requires careful consideration.
    • Centralized Scheduling: A single-threaded central scheduler manages the dependency graph and ensures that the same task instance is not executed by multiple workers simultaneously.

  11. Use dynamic dependencies with yield

    master

    If you cannot determine all dependencies before the task starts, you can use dynamic dependencies. Inside the run() method, you can yield a Task object (or a list of tasks). Luigi will suspend the current task, run the yielded task(s), and then resume the original task.

    Constraints:

    • The run() method must be idempotent because it will resume from the beginning each time a new task is yielded.
    • The yielded task's output (a Target) is returned to the run() method after completion.
    class MyTask(luigi.Task):
    def run(self):
        # The task is suspended here until OtherTask completes
        other_target = yield OtherTask()

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

  12. Understand Luigi's limitations and use cases

    master

    Before implementing Luigi, consider its architectural constraints to ensure it fits your workload:

    • Batch Processing Focus: Luigi is optimized for batch processing. It is not recommended for near real-time pipelines or continuously running processes.
    • Task Granularity: Luigi assumes each task represents a sizable chunk of work. While it can handle thousands of jobs, it is not designed to scale to tens of thousands of tiny tasks.
    • No Distributed Execution: Luigi does not natively distribute task execution across a cluster. A single worker node manages the jobs, which can lead to overloading if not managed (mitigation strategies include triggering from multiple nodes or using resources).
    • No Built-in Triggering: Luigi does not have a built-in scheduler for periodic execution. You must use an external tool like crontab to trigger workflows at specific intervals.