Implementation:Spotify Luigi RunAnywayTarget

Overview

RunAnywayTarget is a Luigi target class in the luigi.contrib.simulate module that enables tasks to be re-executed on every pipeline run. Instead of checking for a persistent output, it uses temporary file markers in the system's temp directory that are scoped to the current process execution. Old markers from previous runs are automatically cleaned up after a configurable time period, ensuring that tasks using this target always appear as incomplete at the start of a new run.

Source Location

Property	Value
Source File	`luigi/contrib/simulate.py`
Lines of Code	111
Module	`luigi.contrib.simulate`
Domain	Testing, Simulation

Import Statement

from luigi.contrib.simulate import RunAnywayTarget

Class: RunAnywayTarget

RunAnywayTarget(luigi.Target)

A target that creates temporary file markers to simulate task completion within a single pipeline execution, allowing tasks to re-run every time the pipeline is invoked.

Class Attributes

Attribute	Type	Default	Description
`temp_dir`	`str`	`os.path.join(tempfile.gettempdir(), 'luigi-simulate')`	The directory where temporary marker files are stored. Subclass this attribute to change the location.
`temp_time`	`int`	`24 * 3600` (86400 seconds / 24 hours)	Maximum age in seconds for marker directories. Directories older than this are deleted on initialization.
`unique`	`multiprocessing.Value('i', 0)`	`0` (initialized to PID on first use)	A process-shared integer value that stores the PID of the first target created in the current execution. This separates marker files between different pipeline runs.

Constructor

RunAnywayTarget.__init__(self, task_obj)

Parameter	Type	Description
`task_obj`	`luigi.Task`	The task instance that owns this target. The task's `task_id` is extracted and used for marker file generation.

The constructor performs two actions:

Extracts task_obj.task_id for later use in path generation.
Sets the unique class-level value to the current PID (using a multiprocessing lock for thread safety) if it has not been set yet.
Cleans up old temporary directories: iterates over entries in temp_dir and deletes any subdirectory whose modification time is older than temp_time seconds.

Methods

Method	Signature	Return Type	Description
`get_path`	`get_path(self)`	`str`	Returns the full path to the temporary marker file. Computed as `{temp_dir}/{unique_pid}/{md5_hash}`, where the MD5 hash is derived from the task's `task_id` string (using `hashlib.new('md5', ..., usedforsecurity=False)`).
`exists`	`exists(self)`	`bool`	Returns `True` if the marker file at `get_path()` exists as a file, `False` otherwise.
`done`	`done(self)`	`None`	Creates the temporary marker file. First ensures the parent directory exists via `os.makedirs()`, then creates an empty file at `get_path()`. This marks the task as "complete" for the current execution.
`__str__`	`__str__(self)`	`str`	Returns the `task_id` string.

Marker File Layout

The temporary marker files follow this directory structure:

{temp_dir}/
  {pid_1}/
    {md5_hash_of_task_id_A}
    {md5_hash_of_task_id_B}
  {pid_2}/
    {md5_hash_of_task_id_C}

Each pipeline execution creates a new subdirectory named by its PID. The temp_time cleanup ensures old PID directories are removed after 24 hours by default.

Usage Example

from luigi.contrib.simulate import RunAnywayTarget
import luigi

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

    def output(self):
        return RunAnywayTarget(self)

    def run(self):
        # Perform the task work
        print("Running task for date: %s" % self.date)

        # Mark task as done for this execution
        self.output().done()

How Re-execution Works

At pipeline start, RunAnywayTarget.exists() returns False because no marker file exists for the current PID.
Luigi schedules the task since its output does not exist.
The task's run() method executes and calls self.output().done().
done() creates the marker file, so exists() now returns True.
If the task is encountered again in the same pipeline run, it is considered complete.
On the next pipeline run, a new PID is used, so the marker file path changes and exists() returns False again.
The cleanup logic in __init__ removes old PID directories after temp_time seconds.

Dependencies

Python standard library: multiprocessing, tempfile, hashlib, logging, os, shutil, time
Luigi core: luigi.Target

Related Principles

Spotify_Luigi_Task_Simulation

Page Connections

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

Principle

Implementation

Heuristic

Environment