Orchestrate Tasks with Luigi

Data pipelines rarely run as one big script. They run as many small steps. Each step depends on another. Some steps are slow. Some fail and need a retry.

Luigi is a Python library that helps you orchestrate these tasks. It tracks dependencies. It handles failures. It shows you a visual overview of your pipeline.

This guide walks you through Luigi from the ground up. You will learn the core concepts, build a working pipeline, and see how Luigi fits into real-world data workflows.

Why Use Luigi for Task Orchestration?

Imagine a pipeline with three steps. First, you download raw data. Then you clean it. Finally, you load it into a database.

If you write this as a plain script, a failure in step two means you rerun everything. That wastes time. It also risks duplicate work.

Luigi solves this with targets. A target is an output that marks a task as done. If the target exists, Luigi skips the task. This makes reruns fast and safe.

Luigi also gives you a central scheduler. It coordinates tasks across workers. It shows progress in a web UI. It retries failed tasks. These features matter when pipelines grow.

Installing Luigi

Luigi is a pure Python package. Install it with pip.

pip install luigi

You can check the version to confirm the install.

import luigi
print(luigi.__version__)

3.5.1

The Core Concepts of Luigi

Luigi has three building blocks. Learn these and you can build almost anything.

Tasks are the units of work. You define them by subclassing luigi.Task. Each task has a run method that does the actual work.

Targets are the outputs of tasks. The output method returns a target. Luigi checks the target to decide if a task is complete.

Dependencies are other tasks that must finish first. The requires method returns them. Luigi builds a graph from these links.

Here is a simple task that writes a file.

import luigi

class WriteGreeting(luigi.Task):
    def output(self):
        # The target marks the task as complete when the file exists
        return luigi.LocalTarget("greeting.txt")

    def run(self):
        # Write the output file
        with self.output().open("w") as f:
            f.write("Hello, Luigi!\n")

Run it from the command line.

python -m luigi --module mytasks WriteGreeting --local-scheduler

===== Luigi Execution Summary =====

Scheduled 1 tasks of which:
* 1 ran successfully:
    - 1 WriteGreeting()

This progress looks :) because there were no failed tasks or missing dependencies

Run it again. Luigi sees the target and skips the task. That is the power of idempotent tasks.

Chaining Tasks with Dependencies

Real pipelines have steps that build on each other. Use requires to link tasks.

The example below reads the greeting file and writes an uppercase version.

import luigi

class WriteGreeting(luigi.Task):
    def output(self):
        return luigi.LocalTarget("greeting.txt")

    def run(self):
        with self.output().open("w") as f:
            f.write("Hello, Luigi!\n")

class ShoutGreeting(luigi.Task):
    def requires(self):
        # This task depends on WriteGreeting
        return WriteGreeting()

    def output(self):
        return luigi.LocalTarget("greeting_upper.txt")

    def run(self):
        # Read the input and write the uppercase output
        with self.input().open("r") as infile:
            text = infile.read().upper()
        with self.output().open("w") as outfile:
            outfile.write(text)

Luigi runs WriteGreeting first. Then it runs ShoutGreeting. You never call them in order yourself. The dependency graph handles it.

This pattern scales well. Add a task that sends an email. Add one that uploads to cloud storage. Each task only knows its direct inputs.

Adding Parameters to Tasks

Hardcoded paths are brittle. Use luigi.Parameter to make tasks configurable.

import luigi

class ProcessFile(luigi.Task):
    # Parameters let you configure the task at runtime
    filename = luigi.Parameter()
    uppercase = luigi.BoolParameter(default=False)

    def output(self):
        suffix = "_upper" if self.uppercase else "_clean"
        return luigi.LocalTarget(f"{self.filename}{suffix}.txt")

    def run(self):
        with open(self.filename, "r") as f:
            text = f.read()
        if self.uppercase:
            text = text.upper()
        with self.output().open("w") as f:
            f.write(text)

Pass parameters on the command line.

python -m luigi --module mytasks ProcessFile --filename data.txt --uppercase --local-scheduler

The parameters become part of the task identity. Two runs with different parameters are different tasks. Luigi tracks them separately.

Handling Failures and Retries

Pipelines fail. Networks drop. Files go missing. Luigi handles this with retries and clear errors.

Set retry_count and retry_delay on a task. Luigi retries failed runs automatically.

class FlakyTask(luigi.Task):
    retry_count = 3
    retry_delay = 10  # seconds between retries

    def output(self):
        return luigi.LocalTarget("flaky_output.txt")

    def run(self):
        # Simulate work that might fail
        with self.output().open("w") as f:
            f.write("done\n")

If a task fails, its dependents do not run. This prevents bad data from spreading. The scheduler marks the failed task and shows it in the UI.

You can also raise luigi.ExternalTask for inputs you do not control. This is useful when the pipeline starts from an existing file or database table.

Using the Luigi Scheduler and Web UI

The local scheduler is fine for testing. For real work, use the central scheduler.

Start it with one command.

luigid

This starts a server on port 8082. Open the browser to see the dashboard. It shows running tasks, completed tasks, and failures.

Workers connect to the scheduler by default. You can run many workers in parallel. Each worker pulls tasks from the queue. Luigi ensures no task runs twice for the same parameters.

The scheduler also stores state. If a worker crashes, the scheduler knows which tasks finished. Restart the worker and it picks up where it left off.

Real-World Use Cases

Luigi shines in data engineering. Teams use it to build ETL pipelines. A task extracts data from an API. Another validates it. A third loads it into a warehouse.

It also fits well with external services. For example, you might wrap API calls in tasks, similar to how you structure OpenAI function calling in Python when building tool-using pipelines. Each call becomes a task with its own target.

Text processing pipelines are another good fit. A task cleans text. Another tokenizes it. If you already work with tokenization with NLTK in Python, you can wrap that logic in a Luigi task and chain it with downstream steps.

Pipelines that move data between systems also benefit. You can write a task that produces messages to a broker, much like you produce and consume Kafka messages in Python, and another task that reads and stores results.

Luigi does not care what the task does. It only cares about inputs, outputs, and dependencies. That makes it flexible across many domains.

Best Practices for Luigi Pipelines

Keep tasks small. One task should do one thing. Small tasks are easier to test and retry.

Make tasks idempotent. Running a task twice should give the same result. Use atomic writes when possible.

Use meaningful targets. A target should tell you the task succeeded. Avoid targets that exist even when the task failed.

Group related tasks in modules. This makes the command line simpler and the code easier to navigate.

Log inside tasks. Luigi captures logs and shows them in the UI. Good logs save hours of debugging.

Luigi vs Other Orchestrators

Luigi is simple and lightweight. It has no database requirement beyond the scheduler state. It runs on a laptop or a cluster.

Airflow is heavier. It uses a database and a web server. It offers more scheduling features like cron and backfills. For many teams, Luigi is enough.

Prefect and Dagster are newer. They offer more dynamic workflows. They also add more dependencies.

Pick Luigi when you want a small, dependable tool. Pick it when your pipeline is a DAG of file-based or API-based steps. Pick it when you value clarity over features.

Common Pitfalls to Avoid

Do not put heavy logic in requires. It should only return task objects. Keep the work in run.

Do not ignore the target check. If your task writes to a path but the target points elsewhere, Luigi will rerun it forever.

Do not share mutable state between tasks. Each task should read its input and write its output. Nothing more.

Do not skip error handling in external calls. Wrap network calls and raise clear exceptions. Luigi will retry, but good errors help humans too.

Conclusion

Luigi turns messy scripts into clear pipelines. Tasks define the work. Targets define success. Dependencies define the order.

Start small. Write one task. Add a target. Chain a second task with requires. Run it with the local scheduler. Then move to luigid when you need scale.

With a few patterns, you can orchestrate tasks that are reliable, rerunnable, and easy to monitor. That is exactly what production pipelines need.