luigi.scheduler module

The system for scheduling tasks and executing them in order. Deals with dependencies, priorities, resources, etc. The Worker pulls tasks from the scheduler (usually over the REST interface) and executes them. See Using the Central Scheduler for more info.

luigi.scheduler.UPSTREAM_SEVERITY_KEY()

T.index(value, [start, [stop]]) -> integer – return first index of value. Raises ValueError if the value is not present.

class luigi.scheduler.RetryPolicy(retry_count, disable_hard_timeout, disable_window)

Bases: tuple

Create new instance of RetryPolicy(retry_count, disable_hard_timeout, disable_window)

disable_hard_timeout

Alias for field number 1

disable_window

Alias for field number 2

retry_count

Alias for field number 0

luigi.scheduler.rpc_method(**request_args)

class luigi.scheduler.scheduler(*args, **kwargs)

Bases: luigi.task.Config

retry_delay = FloatParameter (defaults to 900.0)

remove_delay = FloatParameter (defaults to 600.0)

worker_disconnect_delay = FloatParameter (defaults to 60.0)

state_path = Parameter (defaults to /var/lib/luigi-server/state.pickle)

batch_emails = BoolParameter (defaults to False): Send e-mails in batches rather than immediately

disable_window = IntParameter (defaults to 3600)

retry_count = IntParameter (defaults to 999999999)

disable_hard_timeout = IntParameter (defaults to 999999999)

disable_persist = IntParameter (defaults to 86400)

max_shown_tasks = IntParameter (defaults to 100000)

max_graph_nodes = IntParameter (defaults to 100000)

record_task_history = BoolParameter (defaults to False)

prune_on_get_work = BoolParameter (defaults to False)

pause_enabled = BoolParameter (defaults to True)

send_messages = BoolParameter (defaults to True)

metrics_collector = EnumParameter (defaults to MetricsCollectors.default)

stable_done_cooldown_secs = IntParameter (defaults to 10): Sets cooldown period to avoid running the same task twice

Sets a cooldown period in seconds after a task was completed, during this period the same task will not accepted by the scheduler.

class luigi.scheduler.Failures(window)

Bases: object

This class tracks the number of failures in a given time window.

Failures added are marked with the current timestamp, and this class counts the number of failures in a sliding time window ending at the present.

Initialize with the given window.

Parameters:window – how long to track failures for, as a float (number of seconds).

add_failure()

Add a failure event with the current timestamp.

num_failures()

Return the number of failures in the window.

clear()

Clear the failure queue.

class luigi.scheduler.OrderedSet(iterable=None)

Bases: _abcoll.MutableSet

Standard Python OrderedSet recipe found at http://code.activestate.com/recipes/576694/

Modified to include a peek function to get the last element

add(key)

Add an element.

discard(key)

Remove an element. Do not raise an exception if absent.

peek(last=True)

pop(last=True)

Return the popped value. Raise KeyError if empty.

class luigi.scheduler.Task(task_id, status, deps, resources=None, priority=0, family='', module=None, params=None, param_visibilities=None, accepts_messages=False, tracking_url=None, status_message=None, progress_percentage=None, retry_policy='notoptional')

Bases: object

set_params(params)

is_batchable()

add_failure()

has_excessive_failures()

pretty_id

class luigi.scheduler.Worker(worker_id, last_active=None)

Bases: object

Structure for tracking worker activity and keeping their references.

add_info(info)

update(worker_reference, get_work=False)

prune(config)

get_tasks(state, *statuses)

is_trivial_worker(state)

If it’s not an assistant having only tasks that are without requirements.

We have to pass the state parameter for optimization reasons.

assistant

enabled

state

add_rpc_message(name, **kwargs)

fetch_rpc_messages()

class luigi.scheduler.SimpleTaskState(state_path)

Bases: object

Keep track of the current state and handle persistence.

The point of this class is to enable other ways to keep state, eg. by using a database These will be implemented by creating an abstract base class that this and other classes inherit from.

get_state()

set_state(state)

dump()

load()

get_active_tasks()

get_active_tasks_by_status(*statuses)

get_active_task_count_for_status(status)

get_batch_running_tasks(batch_id)

set_batcher(worker_id, family, batcher_args, max_batch_size)

get_batcher(worker_id, family)

num_pending_tasks()

Return how many tasks are PENDING + RUNNING. O(1).

get_task(task_id, default=None, setdefault=None)

has_task(task_id)

re_enable(task, config=None)

set_batch_running(task, batch_id, worker_id)

set_status(task, new_status, config=None)

fail_dead_worker_task(task, config, assistants)

update_status(task, config)

may_prune(task)

inactivate_tasks(delete_tasks)

get_active_workers(last_active_lt=None, last_get_work_gt=None)

get_assistants(last_active_lt=None)

get_worker_ids()

get_worker(worker_id)

inactivate_workers(delete_workers)

disable_workers(worker_ids)

update_metrics(task, config)

class luigi.scheduler.Scheduler(config=None, resources=None, task_history_impl=None, **kwargs)

Bases: object

Async scheduler that can handle multiple workers, etc.

Can be run locally or on a server (using RemoteScheduler + server.Server).

Keyword Arguments: :param config: an object of class “scheduler” or None (in which the global instance will be used) :param resources: a dict of str->int constraints :param task_history_impl: ignore config and use this object as the task history

load()

dump()

prune()

add_task_batcher(worker, task_family, batched_args, max_batch_size=inf)

forgive_failures(task_id=None)

mark_as_done(task_id=None)

add_task(task_id=None, status='PENDING', runnable=True, deps=None, new_deps=None, expl=None, resources=None, priority=0, family='', module=None, params=None, param_visibilities=None, accepts_messages=False, assistant=False, tracking_url=None, worker=None, batchable=None, batch_id=None, retry_policy_dict=None, owners=None, **kwargs)

• add task identified by task_id if it doesn’t exist
• if deps is not None, update dependency list
• update status of task
• add additional workers/stakeholders
• update priority when needed

announce_scheduling_failure(task_name, family, params, expl, owners, **kwargs)

add_worker(worker, info, **kwargs)

disable_worker(worker)

set_worker_processes(worker, n)

send_scheduler_message(worker, task, content)

add_scheduler_message_response(task_id, message_id, response)

get_scheduler_message_response(task_id, message_id)

has_task_history()

is_pause_enabled()

is_paused()

pause()

unpause()

update_resources(**resources)

update_resource(resource, amount)

count_pending(worker)

get_work(host=None, assistant=False, current_tasks=None, worker=None, **kwargs)

ping(**kwargs)

graph(**kwargs)

dep_graph(task_id, include_done=True, **kwargs)

inverse_dep_graph(task_id, include_done=True, **kwargs)

task_list(status='', upstream_status='', limit=True, search=None, max_shown_tasks=None, **kwargs)

Query for a subset of tasks by status.

worker_list(include_running=True, **kwargs)

resource_list()

Resources usage info and their consumers (tasks).

resources()

get total resources and available ones

task_search(task_str, **kwargs)

Query for a subset of tasks by task_id.

Parameters:task_str – Returns:

re_enable_task(task_id)

fetch_error(task_id, **kwargs)

set_task_status_message(task_id, status_message)

get_task_status_message(task_id)

set_task_progress_percentage(task_id, progress_percentage)

get_task_progress_percentage(task_id)

decrease_running_task_resources(task_id, decrease_resources)

get_running_task_resources(task_id)

task_history

update_metrics_task_started(task)