1. Project Overview & Quickstart (spotify/luigi)
File: doc/central_scheduler.rst
Using the Central Scheduler
While the --local-scheduler flag is useful for development purposes,
it's not recommended for production usage.
The centralized scheduler serves two purposes:
- Make sure two instances of the same task are not running simultaneously
- Provide visualization of everything that's going on.
Note that the central scheduler does not execute anything for you or
help you with job parallelization.
For running tasks periodically,
the easiest thing to do is to trigger a Python script from cron or
from a continuously running process.
There is no central process that automatically triggers jobs.
This model may seem limited, but
we believe that it makes things far more intuitive and easy to understand.
.. figure:: dependency_graph.png
:alt: Dependency graph in the visualiser
The luigid server
To run the server as a daemon run:
.. code-block:: console
$ luigid --background --pidfile --logdir --state-path
Note that this requires ``python-daemon``.
By default, the server starts on AF_INET and AF_INET6 port ``8082``
(which can be changed with the ``--port`` flag) and listens on all IPs. To change the default behavior of listening on all IPs, pass the ``--address`` flag and the IP address to listen on.
To use an AF_UNIX socket use the ``--unix-socket`` flag.
For a full list of configuration options and defaults,
see the :ref:`scheduler configuration section <scheduler-config>`.
Note that ``luigid`` uses the same configuration files as the Luigi client
(i.e. ``luigi.cfg`` or ``/etc/luigi/client.cfg`` by default).
.. _TaskHistory:
Enabling Task HistoryTask History is an experimental feature in which
additional information about tasks that have been executed are recorded in a relational database
for historical analysis.
This information is exposed via the Central Scheduler at /history.
To enable the task history,
specify record_task_history = True in the[scheduler] section of luigi.cfg and
specify db_connection under [task_history].
The db_connection string is used to configure the SQLAlchemy engine <http://docs.sqlalchemy.org/en/rel_0_9/core/engines.html>.
When starting up,luigid will create all the necessary tables using create_all <http://docs.sqlalchemy.org/en/rel_0_9/core/metadata.html#sqlalchemy.schema.MetaData.create_all>.
Example configuration
.. code:: ini
[scheduler]
record_task_history = True
state_path = /usr/local/var/luigi-state.pickle
[task_history]
db_connection = sqlite:////usr/local/var/luigi-task-hist.dbThe task history has the following pages:
/history
a reverse-cronological listing of runs from the past 24 hours.
Example screenshot:.. figure:: history.png
:alt: Recent history screenshot/history/by_id/{id}
detailed information about a run, including:
parameter values, the host on which it ran, and timing information.
Example screenshot:.. figure:: history_by_id.png
:alt: By id screenshot/history/by_name/{name}
a listing of all runs of a task with the given task{name}.
Example screenshot:.. figure:: history_by_name.png
:alt: By name screenshot/history/by_params/{name}?data=params
a listing of all runs of the task{name}restricted to runs withparamsmatching the given history.
Theparamsis a json blob describing the parameters,
e.g.data={"foo": "bar"}looks for a task withfoo=bar./history/by_task_id/{task_id}
the latest run of a task given the{task_id}. It is different from just{id}
and is a derivative ofparams. It is available via{task_id}property of aluigi.Taskinstance or vialuigi.task.task_id_str <https://luigi.readthedocs.io/en/stable/api/luigi.task.html#luigi.task.task_id_str>_.
This kind of representation is useful for concisely recording URLs in a history tree.
Example screenshot:.. figure:: history_by_task_id.png
:alt: By task_id screenshot
File: doc/configuration.rst
Configuration
All configuration can be done by adding configuration files.
Supported config parsers:
cfg(default), based on Python's standard ConfigParser_. Values may refer to environment variables using${ENVVAR}syntax.toml
.. _ConfigParser: https://docs.python.org/3/library/configparser.html
You can choose right parser via LUIGI_CONFIG_PARSER environment variable. For example, LUIGI_CONFIG_PARSER=toml.
Default (cfg) parser are looked for in:
/etc/luigi/client.cfg(deprecated)/etc/luigi/luigi.cfgclient.cfg(deprecated)luigi.cfgLUIGI_CONFIG_PATHenvironment variable
TOML <https://github.com/toml-lang/toml>_ parser are looked for in:
/etc/luigi/luigi.tomlluigi.tomlLUIGI_CONFIG_PATHenvironment variable
Both config lists increase in priority (from low to high). The order only
matters in case of key conflicts (see docs for ConfigParser.read_).
These files are meant for both the client and luigid.
If you decide to specify your own configuration you should make sure
that both the client and luigid load it properly.
.. _ConfigParser.read: https://docs.python.org/3/library/configparser.html#configparser.ConfigParser.read
The config file is broken into sections, each controlling a different part of the config.
Example cfg config:
.. code:: ini
[hadoop]
version=cdh4
streaming_jar=/usr/lib/hadoop-xyz/hadoop-streaming-xyz-123.jar
[core]
scheduler_host=luigi-host.mycompany.fooExample toml config:
.. code:: python
[hadoop]
version = "cdh4"
streaming_jar = "/usr/lib/hadoop-xyz/hadoop-streaming-xyz-123.jar"
[core]
scheduler_host = "luigi-host.mycompany.foo"Also see examples/config.toml <https://github.com/spotify/luigi/blob/master/examples/config.toml>_
for more complex example.
.. _ParamConfigIngestion:
Parameters from config Ingestion
All parameters can be overridden from configuration files. For instance if you
have a Task definition:
.. code:: python
class DailyReport(luigi.contrib.hadoop.JobTask):
date = luigi.DateParameter(default=datetime.date.today())
# ...Then you can override the default value for DailyReport().date by providing
it in the configuration:
.. code:: ini
[DailyReport]
date=2012-01-01.. _ConfigClasses:
Configuration classes
Using the :ref:ParamConfigIngestion method, we derive the
conventional way to do global configuration. Imagine this configuration.
.. code:: ini
[mysection]
option=hello
intoption=123We can create a :py:class:~luigi.Config class:
.. code:: python
import luigi
# Config classes should be camel cased
class mysection(luigi.Config):
option = luigi.Parameter(default='world')
intoption = luigi.IntParameter(default=555)
mysection().option
mysection().intoptionConfigurable options
Luigi comes with a lot of configurable options. Below, we describe each
section and the parameters available within it.
[core]
These parameters control core Luigi behavior, such as error e-mails and
interactions between the worker and scheduler.
autoload_range
.. versionadded:: 2.8.11
If false, prevents range tasks from autoloading. They can still be loaded
using --module luigi.tools.range. Defaults to true. Setting this to true
explicitly disables the deprecation warning.
default_scheduler_host
Hostname of the machine running the scheduler. Defaults to localhost.
default_scheduler_port
Port of the remote scheduler api process. Defaults to 8082.
default_scheduler_url
Full path to remote scheduler. Defaults to http://localhost:8082/.
For TLS support use the URL scheme: https,
example: https://luigi.example.com:443/
(Note: you will have to terminate TLS using an HTTP proxy)
You can also use this to connect to a local Unix socket using the
non-standard URI scheme: http+unix
example: http+unix://%2Fvar%2Frun%2Fluigid%2Fluigid.sock/
hdfs_tmp_dir
Base directory in which to store temporary files on hdfs. Defaults to
tempfile.gettempdir()
history_filename
If set, specifies a filename for Luigi to write stuff (currently just
job id) to in mapreduce job's output directory. Useful in a
configuration where no history is stored in the output directory by
Hadoop.
log_level
The default log level to use when no logging_conf_file is set. Must be
a valid name of a Python log level <https://docs.python.org/2/library/logging.html#logging-levels>_.
Default is DEBUG.
logging_conf_file
Location of the logging configuration file.
no_configure_logging
If true, logging is not configured. Defaults to false.
parallel_scheduling
If true, the scheduler will compute complete functions of tasks in
parallel using multiprocessing. This can significantly speed up
scheduling, but requires that all tasks can be pickled.
Defaults to false.
parallel_scheduling_processes
The number of processes to use for parallel scheduling. If not specified
the default number of processes will be the total number of CPUs available.
rpc_connect_timeout
Number of seconds to wait before timing out when making an API call.
Defaults to 10.0
rpc_retry_attempts
The maximum number of retries to connect the central scheduler before giving up.
Defaults to 3
rpc_retry_wait
Number of seconds to wait before the next attempt will be started to
connect to the central scheduler between two retry attempts.
Defaults to 30
[cors]
.. versionadded:: 2.8.0
These parameters control /api/<method> CORS behaviour (see: W3C Cross-Origin Resource Sharing <http://www.w3.org/TR/cors/>_).
enabled
Enables CORS support.
Defaults to false.
allowed_origins
A list of allowed origins. Used only if allow_any_origin is false.
Configure in JSON array format, e.g. ["foo", "bar"].
Defaults to empty.
allow_any_origin
Accepts requests from any origin.
Defaults to false.
allow_null_origin
Allows the request to set null value of the Origin header.
Defaults to false.
max_age
Content of Access-Control-Max-Age.
Defaults to 86400 (24 hours).
allowed_methods
Content of Access-Control-Allow-Methods.
Defaults to GET, OPTIONS.
allowed_headers
Content of Access-Control-Allow-Headers.
Defaults to Accept, Content-Type, Origin.
exposed_headers
Content of Access-Control-Expose-Headers.
Defaults to empty string (will NOT be sent as a response header).
allow_credentials
Indicates that the actual request can include user credentials.
Defaults to false.
.. _worker-config:
[worker]
These parameters control Luigi worker behavior.
count_uniques
If true, workers will only count unique pending jobs when deciding
whether to stay alive. So if a worker can't get a job to run and other
workers are waiting on all of its pending jobs, the worker will die.
worker_keep_alive must be true for this to have any effect. Defaults
to false.
keep_alive
If true, workers will stay alive when they run out of jobs to run, as
long as they have some pending job waiting to be run. Defaults to
false.
ping_interval
Number of seconds to wait between pinging scheduler to let it know
that the worker is still alive. Defaults to 1.0.
task_limit
.. versionadded:: 1.0.25
Maximum number of tasks to schedule per invocation. Upon exceeding it,
the worker will issue a warning and proceed with the workflow obtained
thus far. Prevents incidents due to spamming of the scheduler, usually
accidental. Default: no limit.
task_process_context
An optional setting allowing Luigi to import a custom context manager
used to wrap the execution of tasks' run methods. Default: no context manager.
timeout
.. versionadded:: 1.0.20
Number of seconds after which to kill a task which has been running
for too long. This provides a default value for all tasks, which can
be overridden by setting the worker_timeout property in any task.
Default value is 0, meaning no timeout.
wait_interval
Number of seconds for the worker to wait before asking the scheduler
for another job after the scheduler has said that it does not have any
available jobs.
wait_jitter
Duration of jitter to add to the worker wait interval such that the multiple
workers do not ask the scheduler for another job at the same time, in seconds.
Default: 5.0
max_keep_alive_idle_duration
.. versionadded:: 2.8.4
Maximum duration in seconds to keep worker alive while in idle state.
Default: 0 (Indefinitely)
max_reschedules
The maximum number of times that a job can be automatically
rescheduled by a worker before it will stop trying. Workers will
reschedule a job if it is found to not be done when attempting to run
a dependent job. This defaults to 1.
retry_external_tasks
If true, incomplete external tasks (i.e. tasks where the run() method is
NotImplemented) will be retested for completion while Luigi is running.
This means that if external dependencies are satisfied after a workflow has
started, any tasks dependent on that resource will be eligible for running.
Note: Every time the task remains incomplete, it will count as FAILED, so
normal retry logic applies (see: retry_count and retry_delay).
This setting works best with worker_keep_alive: true.
If false, external tasks will only be evaluated when Luigi is first invoked.
In this case, Luigi will not check whether external dependencies are
satisfied while a workflow is in progress, so dependent tasks will remain
PENDING until the workflow is reinvoked.
Defaults to false for backwards compatibility.
no_install_shutdown_handler
By default, workers will stop requesting new work and finish running
pending tasks after receiving a SIGUSR1 signal. This provides a hook
for gracefully shutting down workers that are in the process of running
(potentially expensive) tasks. If set to true, Luigi will NOT install
this shutdown hook on workers. Note this hook does not work on Windows
operating systems, or when jobs are launched outside the main execution
thread.
Defaults to false.
send_failure_email
Controls whether the worker will send e-mails on task and scheduling
failures. If set to false, workers will only send e-mails on
framework errors during scheduling and all other e-mail must be
handled by the scheduler.
Defaults to true.
check_unfulfilled_deps
If true, the worker checks for completeness of dependencies before running a
task. In case unfulfilled dependencies are detected, an exception is raised
and the task will not run. This mechanism is useful to detect situations
where tasks do not create their outputs properly, or when targets were
removed after the dependency tree was built. It is recommended to disable
this feature only when the completeness checks are known to be bottlenecks,
e.g. when the exists() calls of the dependencies' outputs are
resource-intensive.
Defaults to true.
force_multiprocessing
By default, luigi uses multiprocessing when more than one worker process is
requested. When set to true, multiprocessing is used independent of the
number of workers.
Defaults to false.
check_complete_on_run
By default, luigi tasks are marked as 'done' when they finish running without
raising an error. When set to true, tasks will also verify that their outputs
exist when they finish running, and will fail immediately if the outputs are
missing.
Defaults to false.
cache_task_completion
By default, luigi task processes might check the completion status multiple
times per task which is a safe way to avoid potential inconsistencies. For
tasks with many dynamic dependencies, yielded in multiple stages, this might
become expensive, e.g. in case the per-task completion check entails remote
resources. When set to true, completion checks are cached so that tasks
declared as complete once are not checked again.
Defaults to false.
[elasticsearch]
These parameters control use of elasticsearch
marker_index
Defaults to "update_log".
marker_doc_type
Defaults to "entry".
[email]
General parameters
force_send
If true, e-mails are sent in all run configurations (even if stdout is
connected to a tty device). Defaults to False.
format
Type of e-mail to send. Valid values are "plain", "html" and "none".
When set to html, tracebacks are wrapped in tags to get fixed-
width font. When set to none, no e-mails will be sent.
Default value is plain.
method
Valid values are "smtp", "sendgrid", "ses" and "sns". SES and SNS are
services of Amazon web services. SendGrid is an email delivery service.
The default value is "smtp".
In order to send messages through Amazon SNS or SES set up your AWS
config files or run Luigi on an EC2 instance with proper instance
profile.
In order to use sendgrid, fill in your sendgrid API key in the
[sendgrid]_ section.
In order to use smtp, fill in the appropriate fields in the [smtp]_
section.
prefix
Optional prefix to add to the subject line of all e-mails. For
example, setting this to "[LUIGI]" would change the subject line of an
e-mail from "Luigi: Framework error" to "[LUIGI] Luigi: Framework
error"
receiver
Recipient of all error e-mails. If this is not set, no error e-mails
are sent when Luigi crashes unless the crashed job has owners set. If
Luigi is run from the command line, no e-mails will be sent unless
output is redirected to a file.
Set it to SNS Topic ARN if you want to receive notifications through
Amazon SNS. Make sure to set method to sns in this case too.
sender
User name in from field of error e-mails.
Default value: luigi-client@
traceback_max_length
Maximum length for traceback included in error email. Default is 5000.
[batch_email]
Parameters controlling the contents of batch notifications sent from the
scheduler
email_interval
Number of minutes between e-mail sends. Making this larger results in
fewer, bigger e-mails.
Defaults to 60.
batch_mode
Controls how tasks are grouped together in the e-mail. Suppose we have
the following sequence of failures:
- TaskA(a=1, b=1)
- TaskA(a=1, b=1)
- TaskA(a=2, b=1)
- TaskA(a=1, b=2)
- TaskB(a=1, b=1)
For any setting of batch_mode, the batch e-mail will record 5 failures
and mention them in the subject. The difference is in how they will
be displayed in the body. Here are example bodies with error_messages
set to 0.
"all" only groups together failures for the exact same task:
- TaskA(a=1, b=1) (2 failures)
- TaskA(a=1, b=2) (1 failure)
- TaskA(a=2, b=1) (1 failure)
- TaskB(a=1, b=1) (1 failure)
"family" groups together failures for tasks of the same family:
- TaskA (4 failures)
- TaskB (1 failure)
"unbatched_params" groups together tasks that look the same after
removing batched parameters. So if TaskA has a batch_method set for
parameter a, we get the following:
- TaskA(b=1) (3 failures)
- TaskA(b=2) (1 failure)
- TaskB(a=1, b=2) (1 failure)
Defaults to "unbatched_params", which is identical to "all" if you are
not using batched parameters.
error_lines
Number of lines to include from each error message in the batch
e-mail. This can be used to keep e-mails shorter while preserving the
more useful information usually found near the bottom of stack traces.
This can be set to 0 to include all lines. If you don't wish to see
error messages, instead set error_messages to 0.
Defaults to 20.
error_messages
Number of messages to preserve for each task group. As most tasks that
fail repeatedly do so for similar reasons each time, it's not usually
necessary to keep every message. This controls how many messages are
kept for each task or task group. The most recent error messages are
kept. Set to 0 to not include error messages in the e-mails.
Defaults to 1.
group_by_error_messages
Quite often, a system or cluster failure will cause many disparate
task types to fail for the same reason. This can cause a lot of noise
in the batch e-mails. This cuts down on the noise by listing items
with identical error messages together. Error messages are compared
after limiting by error_lines.
Defaults to true.
[hadoop]
Parameters controlling basic hadoop tasks
command
Name of command for running hadoop from the command line. Defaults to
"hadoop"
python_executable
Name of command for running python from the command line. Defaults to
"python"
scheduler
Type of scheduler to use when scheduling hadoop jobs. Can be "fair" or
"capacity". Defaults to "fair".
streaming_jar
Path to your streaming jar. Must be specified to run streaming jobs.
version
Version of hadoop used in your cluster. Can be "cdh3", "chd4", or
"apache1". Defaults to "cdh4".
[hdfs]
Parameters controlling the use of snakebite to speed up hdfs queries.
client
Client to use for most hadoop commands. Options are "snakebite",
"snakebite_with_hadoopcli_fallback", "webhdfs" and "hadoopcli". Snakebite is
much faster, so use of it is encouraged. webhdfs is fast and works with
Python 3 as well, but has not been used that much in the wild.
Both snakebite and webhdfs requires you to install it separately on
the machine. Defaults to "hadoopcli".
client_version
Optionally specifies hadoop client version for snakebite.
effective_user
Optionally specifies the effective user for snakebite.
namenode_host
The hostname of the namenode. Needed for snakebite if
snakebite_autoconfig is not set.
namenode_port
The port used by snakebite on the namenode. Needed for snakebite if
snakebite_autoconfig is not set.
snakebite_autoconfig
If true, attempts to automatically detect the host and port of the
namenode for snakebite queries. Defaults to false.
tmp_dir
Path to where Luigi will put temporary files on hdfs
[hive]
Parameters controlling hive tasks
command
Name of the command used to run hive on the command line. Defaults to
"hive".
hiverc_location
Optional path to hive rc file.
metastore_host
Hostname for metastore.
metastore_port
Port for hive to connect to metastore host.
release
If set to "apache", uses a hive client that better handles apache
hive output. All other values use the standard client Defaults to
"cdh4".
[kubernetes]
Parameters controlling Kubernetes Job Tasks
auth_method
Authorization method to access the cluster.
Options are "kubeconfig_" or "service-account_"
kubeconfig_path
Path to kubeconfig file, for cluster authentication.
It defaults to ~/.kube/config, which is the default location when
using minikube_.
When auth_method is "service-account" this property is ignored.
max_retrials
Maximum number of retrials in case of job failure.
.. _service-account: http://kubernetes.io/docs/user-guide/kubeconfig-file
.. _kubeconfig: http://kubernetes.io/docs/user-guide/service-accounts
.. _minikube: http://kubernetes.io/docs/getting-started-guides/minikube
[mysql]
Parameters controlling use of MySQL targets
marker_table
Table in which to store status of table updates. This table will be
created if it doesn't already exist. Defaults to "table_updates".
[postgres]
Parameters controlling the use of Postgres targets
local_tmp_dir
Directory in which to temporarily store data before writing to
postgres. Uses system default if not specified.
marker_table
Table in which to store status of table updates. This table will be
created if it doesn't already exist. Defaults to "table_updates".
[prometheus]
use_task_family_in_labels
Should task family be used as a prometheus bucket label.
Default value is true.
task_parameters_to_use_in_labels
List of task arguments' names used as additional prometheus bucket labels.
Passed in a form of a json list.
[redshift]
Parameters controlling the use of Redshift targets
marker_table
Table in which to store status of table updates. This table will be
created if it doesn't already exist. Defaults to "table_updates".
.. _resources-config:
[resources]
This section can contain arbitrary keys. Each of these specifies the
amount of a global resource that the scheduler can allow workers to use.
The scheduler will prevent running jobs with resources specified from
exceeding the counts in this section. Unspecified resources are assumed
to have limit 1. Example resources section for a configuration with 2
hive resources and 1 mysql resource:
.. code:: ini
[resources]
hive=2
mysql=1
Note that it was not necessary to specify the 1 for mysql here, but it
is good practice to do so when you have a fixed set of resources.
.. _retcode-config:
[retcode]
Configure return codes for the Luigi binary. In the case of multiple return
codes that could apply, for example a failing task and missing data, the
numerically greatest return code is returned.
We recommend that you copy this set of exit codes to your luigi.cfg file:
.. code:: ini
[retcode]
The following return codes are the recommended exit codes for Luigi
They are in increasing level of severity (for most applications)
already_running=10
missing_data=20
not_run=25
task_failed=30
scheduling_error=35
unhandled_exception=40
already_running
This can happen in two different cases. Either the local lock file was taken
at the time the invocation starts up. Or, the central scheduler have reported
that some tasks could not have been run, because other workers are already
running the tasks.
missing_data
For when an :py:class:~luigi.task.ExternalTask is not complete, and this
caused the worker to give up. As an alternative to fiddling with this, see
the [worker] keep_alive option.
not_run
For when a task is not granted run permission by the scheduler. Typically
because of lack of resources, because the task has been already run by
another worker or because the attempted task is in DISABLED state.
Connectivity issues with the central scheduler might also cause this.
This does not include the cases for which a run is not allowed due to missing
dependencies (missing_data) or due to the fact that another worker is currently
running the task (already_running).
task_failed
For signaling that there were last known to have failed. Typically because
some exception have been raised.
scheduling_error
For when a task's complete() or requires() method fails with an
exception, or when the limit number of tasks is reached.
unhandled_exception
For internal Luigi errors. Defaults to 4, since this type of error
probably will not recover over time.
If you customize return codes, prefer to set them in range 128 to 255 to avoid
conflicts. Return codes in range 0 to 127 are reserved for possible future use
by Luigi contributors.
[scalding]
Parameters controlling running of scalding jobs
scala_home
Home directory for scala on your machine. Defaults to either
SCALA_HOME or /usr/share/scala if SCALA_HOME is unset.
scalding_home
Home directory for scalding on your machine. Defaults to either
SCALDING_HOME or /usr/share/scalding if SCALDING_HOME is unset.
scalding_provided
Provided directory for scalding on your machine. Defaults to either
SCALDING_HOME/provided or /usr/share/scalding/provided
scalding_libjars
Libjars directory for scalding on your machine. Defaults to either
SCALDING_HOME/libjars or /usr/share/scalding/libjars
.. _scheduler-config:
[scheduler]
Parameters controlling scheduler behavior
batch_emails
Whether to send batch e-mails for failures and disables rather than
sending immediate disable e-mails and just relying on workers to send
immediate batch e-mails.
Defaults to false.
disable_hard_timeout
Hard time limit after which tasks will be disabled by the server if
they fail again, in seconds. It will disable the task if it fails
again after this amount of time. E.g. if this was set to 600
(i.e. 10 minutes), and the task first failed at 10:00am, the task would
be disabled if it failed again any time after 10:10am. Note: This setting
does not consider the values of the retry_count or
disable_window settings.
retry_count
Number of times a task can fail within disable_window before
the scheduler will automatically disable it. If not set, the scheduler
will not automatically disable jobs.
disable_persist
Number of seconds for which an automatic scheduler disable lasts.
Defaults to 86400 (1 day).
disable_window
Number of seconds during which retry_count failures must
occur in order for an automatic disable by the scheduler. The
scheduler forgets about disables that have occurred longer ago than
this amount of time. Defaults to 3600 (1 hour).
max_shown_tasks
.. versionadded:: 1.0.20
The maximum number of tasks returned in a task_list api call. This
will restrict the number of tasks shown in task lists in the
visualiser. Small values can alleviate frozen browsers when there are
too many done tasks. This defaults to 100000 (one hundred thousand).
max_graph_nodes
.. versionadded:: 2.0.0
The maximum number of nodes returned by a dep_graph or
inverse_dep_graph api call. Small values can greatly speed up graph
display in the visualiser by limiting the number of nodes shown. Some
of the nodes that are not sent to the visualiser will still show up as
dependencies of nodes that were sent. These nodes are given TRUNCATED
status.
record_task_history
If true, stores task history in a database. Defaults to false.
remove_delay
Number of seconds to wait before removing a task that has no
stakeholders. Defaults to 600 (10 minutes).
retry_delay
Number of seconds to wait after a task failure to mark it pending
again. Defaults to 900 (15 minutes).
state_path
Path in which to store the Luigi scheduler's state. When the scheduler
is shut down, its state is stored in this path. The scheduler must be
shut down cleanly for this to work, usually with a kill command. If
the kill command includes the -9 flag, the scheduler will not be able
to save its state. When the scheduler is started, it will load the
state from this path if it exists. This will restore all scheduled
jobs and other state from when the scheduler last shut down.
Sometimes this path must be deleted when restarting the scheduler
after upgrading Luigi, as old state files can become incompatible
with the new scheduler. When this happens, all workers should be
restarted after the scheduler both to become compatible with the
updated code and to reschedule the jobs that the scheduler has now
forgotten about.
This defaults to /var/lib/luigi-server/state.pickle
worker_disconnect_delay
Number of seconds to wait after a worker has stopped pinging the
scheduler before removing it and marking all of its running tasks as
failed. Defaults to 60.
pause_enabled
If false, disables pause/unpause operations and hides the pause toggle from
the visualiser.
send_messages
When true, the scheduler is allowed to send messages to running tasks and
the central scheduler provides a simple prompt per task to send messages.
Defaults to true.
metrics_collector
Optional setting allowing Luigi to use a contribution to collect metrics
about the pipeline to a third-party. By default this uses the default metric
collector that acts as a shell and does nothing. The currently available
options are "datadog", "prometheus" and "custom". If it's custom the
'metrics_custom_import' needs to be set.
metrics_custom_import
Optional setting allowing Luigi to import a custom subclass of MetricsCollector
at runtime. The string should be formatted like "module.sub_module.ClassName".
[sendgrid]
These parameters control sending error e-mails through SendGrid.
apikey
API key of the SendGrid account.
[smtp]
These parameters control the smtp server setup.
host
Hostname for sending mail through smtp. Defaults to localhost.
local_hostname
If specified, overrides the FQDN of localhost in the HELO/EHLO
command.
no_tls
If true, connects to smtp without TLS. Defaults to false.
password
Password to log in to your smtp server. Must be specified for
username to have an effect.
port
Port number for smtp on smtp_host. Defaults to 0.
ssl
If true, connects to smtp through SSL. Defaults to false.
timeout
Sets the number of seconds after which smtp attempts should time out.
Defaults to 10.
username
Username to log in to your smtp server, if necessary.
[spark]
Parameters controlling the default execution of :py:class:~luigi.contrib.spark.SparkSubmitTask and :py:class:~luigi.contrib.spark.PySparkTask:
.. deprecated:: 1.1.1
:py:class:~luigi.contrib.spark.SparkJob, :py:class:~luigi.contrib.spark.Spark1xJob and :py:class:~luigi.contrib.spark.PySpark1xJob
are deprecated. Please use :py:class:~luigi.contrib.spark.SparkSubmitTask or :py:class:~luigi.contrib.spark.PySparkTask.
spark_submit
Command to run in order to submit spark jobs. Default: "spark-submit"
master
Master url to use for spark_submit. Example: local[*], spark://masterhost:7077. Default: Spark default (Prior to 1.1.1: yarn-client)
deploy_mode
Whether to launch the driver programs locally ("client") or on one of the worker machines inside the cluster ("cluster"). Default: Spark default
jars
Comma-separated list of local jars to include on the driver and executor classpaths. Default: Spark default
packages
Comma-separated list of packages to link to on the driver and executors
py_files
Comma-separated list of .zip, .egg, or .py files to place on the PYTHONPATH for Python apps. Default: Spark default
files
Comma-separated list of files to be placed in the working directory of each executor. Default: Spark default
conf:
Arbitrary Spark configuration property in the form Prop=Value|Prop2=Value2. Default: Spark default
properties_file
Path to a file from which to load extra properties. Default: Spark default
driver_memory
Memory for driver (e.g. 1000M, 2G). Default: Spark default
driver_java_options
Extra Java options to pass to the driver. Default: Spark default
driver_library_path
Extra library path entries to pass to the driver. Default: Spark default
driver_class_path
Extra class path entries to pass to the driver. Default: Spark default
executor_memory
Memory per executor (e.g. 1000M, 2G). Default: Spark default
Configuration for Spark submit jobs on Spark standalone with cluster deploy mode only:
driver_cores
Cores for driver. Default: Spark default
supervise
If given, restarts the driver on failure. Default: Spark default
Configuration for Spark submit jobs on Spark standalone and Mesos only:
total_executor_cores
Total cores for all executors. Default: Spark default
Configuration for Spark submit jobs on YARN only:
executor_cores
Number of cores per executor. Default: Spark default
queue
The YARN queue to submit to. Default: Spark default
num_executors
Number of executors to launch. Default: Spark default
archives
Comma separated list of archives to be extracted into the working directory of each executor. Default: Spark default
hadoop_conf_dir
Location of the hadoop conf dir. Sets HADOOP_CONF_DIR environment variable
when running spark. Example: /etc/hadoop/conf
Extra configuration for PySparkTask jobs:
py_packages
Comma-separated list of local packages (in your python path) to be distributed to the cluster.
Parameters controlling the execution of SparkJob jobs (deprecated):
[task_history]
Parameters controlling storage of task history in a database
db_connection
Connection string for connecting to the task history db using
sqlalchemy.
[execution_summary]
Parameters controlling execution summary of a worker
summary_length
Maximum number of tasks to show in an execution summary. If the value is 0,
then all tasks will be displayed. Default value is 5.
[webhdfs]
port
The port to use for webhdfs. The normal namenode port is probably on a
different port from this one.
user
Perform file system operations as the specified user instead of $USER. Since
this parameter is not honored by any of the other hdfs clients, you should
think twice before setting this parameter.
client_type
The type of client to use. Default is the "insecure" client that requires no
authentication. The other option is the "kerberos" client that uses kerberos
authentication.
[datadog]
api_key
The api key found in the account settings of Datadog under the API
sections.
app_key
The application key found in the account settings of Datadog under the API
sections.
default_tags
Optional settings that adds the tag to all the metrics and events sent to
Datadog. Default value is "application:luigi".
environment
Allows you to tweak multiple environment to differentiate between production,
staging or development metrics within Datadog. Default value is "development".
statsd_host
The host that has the statsd instance to allow Datadog to send statsd metric. Default value is "localhost".
statsd_port
The port on the host that allows connection to the statsd host. Defaults value is 8125.
metric_namespace
Optional prefix to add to the beginning of every metric sent to Datadog.
Default value is "luigi".
Per Task Retry-Policy
Luigi also supports defining retry_policy per task.
.. code-block:: python
class GenerateWordsFromHdfs(luigi.Task):
retry_count = 2
...
class GenerateWordsFromRDBM(luigi.Task):
retry_count = 5
...
class CountLetters(luigi.Task):
def requires(self):
return [GenerateWordsFromHdfs()]
def run():
yield GenerateWordsFromRDBM()
...If none of retry-policy fields is defined per task, the field value will be default value which is defined in luigi config file.
To make luigi sticks to the given retry-policy, be sure you run luigi worker with keep_alive config. Please check keep_alive config in :ref:worker-config section.
Retry-Policy Fields
The fields below are in retry-policy and they can be defined per task.
retry_countdisable_hard_timeoutdisable_window
File: doc/design_and_limitations.rst
Design and limitations
Luigi is the successor to a couple of attempts that we weren't fully happy with.
We learned a lot from our mistakes and some design decisions include:
- Straightforward command-line integration.
- As little boilerplate as possible.
- Focus on job scheduling and dependency resolution, not a particular platform.
In particular, this means no limitation to Hadoop.
Though Hadoop/HDFS support is built-in and is easy to use,
this is just one of many types of things you can run. - A file system abstraction where code doesn't have to care about where files are located.
- Atomic file system operations through this abstraction.
If a task crashes it won't lead to a broken state. - The dependencies are decentralized.
No big config file in XML.
Each task just specifies which inputs it needs and cross-module dependencies are trivial. - A web server that renders the dependency graph and does locking, etc for free.
- Trivial to extend with new file systems, file formats, and job types.
You can easily write jobs that inserts a Tokyo Cabinet into Cassandra.
Adding support for new systems is generally not very hard.
(Feel free to send us a patch when you're done!) - Date algebra included.
- Lots of unit tests of the most basic stuff.
It wouldn't be fair not to mention some limitations with the current design:
- Its focus is on batch processing so
it's probably less useful for near real-time pipelines or continuously running processes. - The assumption is that each task is a sizable chunk of work.
While you can probably schedule a few thousand jobs,
it's not meant to scale beyond tens of thousands. - Luigi does not support distribution of execution.
When you have workers running thousands of jobs daily, this starts to matter,
because the worker nodes get overloaded.
There are some ways to mitigate this (trigger from many nodes, use resources),
but none of them are ideal. - Luigi does not come with built-in triggering, and you still need to rely on something like
crontab to trigger workflows periodically.
Also, it should be mentioned that Luigi is named after the world's second most famous plumber.
File: doc/example_top_artists.rst
Example – Top Artists
This is a very simplified case of something we do at Spotify a lot.
All user actions are logged to Google Cloud Storage (previously HDFS) where
we run a bunch of processing jobs to transform the data. The processing code itself is implemented
in a scalable data processing framework, such as Scio, Scalding, or Spark, but the jobs
are orchestrated with Luigi.
At some point we might end up with
a smaller data set that we can bulk ingest into Cassandra, Postgres, or
other storage suitable for serving or exploration.
For the purpose of this exercise, we want to aggregate all streams,
find the top 10 artists and then put the results into Postgres.
This example is also available inexamples/top_artists.py <https://github.com/spotify/luigi/blob/master/examples/top_artists.py>_.
Step 1 - Aggregate Artist Streams
.. code:: python
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):
artist_count = defaultdict(int)
for input in self.input():
with input.open('r') as in_file:
for line in in_file:
timestamp, artist, track = line.strip().split()
artist_count[artist] += 1
with self.output().open('w') as out_file:
for artist, count in artist_count.iteritems():
print(artist, count, file=out_file)
Note that this is just a portion of the file ``examples/top_artists.py``.
In particular, ``Streams`` is defined as a :class:`~luigi.task.Task`,
acting as a dependency for ``AggregateArtists``.
In addition, ``luigi.run()`` is called if the script is executed directly,
allowing it to be run from the command line.
There are several pieces of this snippet that deserve more explanation.
- Any :class:`~luigi.task.Task` may be customized by instantiating one
or more :class:`~luigi.parameter.Parameter` objects on the class level.
- The :func:`~luigi.task.Task.output` method tells Luigi where the result
of running the task will end up. The path can be some function of the
parameters.
- The :func:`~luigi.task.Task.requires` tasks specifies other tasks that
we need to perform this task. In this case it's an external dump named
*Streams* which takes the date as the argument.
- For plain Tasks, the :func:`~luigi.task.Task.run` method implements the
task. This could be anything, including calling subprocesses, performing
long running number crunching, etc. For some subclasses of
:class:`~luigi.task.Task` you don't have to implement the ``run``
method. For instance, for the :class:`~luigi.contrib.hadoop.JobTask`
subclass you implement a *mapper* and *reducer* instead.
- :class:`~luigi.LocalTarget` is a built in class that makes it
easy to read/write from/to the local filesystem. It also makes all file operations
atomic, which is nice in case your script crashes for any reason.
Running this Locally
~~~~~~~~~~~~~~~~~~~~
Try running this using eg.
.. code-block:: console
$ cd examples
$ luigi --module top_artists AggregateArtists --local-scheduler --date-interval 2012-06
Note that *top_artists* needs to be in your PYTHONPATH, or else this can produce an error (*ImportError: No module named top_artists*). Add the current working directory to the command PYTHONPATH with:
.. code-block:: console
$ PYTHONPATH='.' luigi --module top_artists AggregateArtists --local-scheduler --date-interval 2012-06
You can also try to view the manual using ``--help`` which will give you an
overview of the options.
Running the command again will do nothing because the output file is
already created.
In that sense, any task in Luigi is *idempotent*
because running it many times gives the same outcome as running it once.
Note that unlike Makefile, the output will not be recreated when any of
the input files is modified.
You need to delete the output file
manually.
The ``--local-scheduler`` flag tells Luigi not to connect to a scheduler
server. This is not recommended for other purpose than just testing
things.
Step 1b - Aggregate artists with SparkWhile Luigi can process data inline, it is normally used to orchestrate external programs that
perform the actual processing. In this example, we will demonstrate how top artists instead can be
read from HDFS and calculated with Spark, orchestrated by Luigi.
.. code:: python
class AggregateArtistsSpark(luigi.contrib.spark.SparkSubmitTask):
date_interval = luigi.DateIntervalParameter()
app = 'top_artists_spark.py'
master = 'local[*]'
def output(self):
return luigi.contrib.hdfs.HdfsTarget("data/artist_streams_%s.tsv" % self.date_interval)
def requires(self):
return [StreamsHdfs(date) for date in self.date_interval]
def app_options(self):
# :func:`~luigi.task.Task.input` returns the targets produced by the tasks in
# `~luigi.task.Task.requires`.
return [','.join([p.path for p in self.input()]),
self.output().path]:class:luigi.contrib.hadoop.SparkSubmitTask doesn't require you to implement a
:func:~luigi.task.Task.run method. Instead, you specify the command line parameters to send
to spark-submit, as well as any other configuration specific to Spark.
Python code for the Spark job is found below.
.. code:: python
import operator
import sys
from pyspark.sql import SparkSession
def main(argv):
input_paths = argv[1].split(',')
output_path = argv[2]
spark = SparkSession.builder.getOrCreate()
streams = spark.read.option('sep', '\t').csv(input_paths[0])
for stream_path in input_paths[1:]:
streams.union(spark.read.option('sep', '\t').csv(stream_path))
# The second field is the artist
counts = streams \
.map(lambda row: (row[1], 1)) \
.reduceByKey(operator.add)
counts.write.option('sep', '\t').csv(output_path)
if __name__ == '__main__':
sys.exit(main(sys.argv))In a typical deployment scenario, the Luigi orchestration definition above as well as the
Pyspark processing code would be packaged into a deployment package, such as a container image. The
processing code does not have to be implemented in Python, any program can be packaged in the
image and run from Luigi.
Step 2 – Find the Top Artists
At this point, we've counted the number of streams for each artists,
for the full time period.
We are left with a large file that contains
mappings of artist -> count data, and we want to find the top 10 artists.
Since we only have a few hundred thousand artists, and
calculating artists is nontrivial to parallelize,
we choose to do this not as a Hadoop job, but just as a plain old for-loop in Python.
.. code:: python
class Top10Artists(luigi.Task):
date_interval = luigi.DateIntervalParameter()
use_hadoop = luigi.BoolParameter()
def requires(self):
if self.use_hadoop:
return AggregateArtistsSpark(self.date_interval)
else:
return AggregateArtists(self.date_interval)
def output(self):
return luigi.LocalTarget("data/top_artists_%s.tsv" % self.date_interval)
def run(self):
top_10 = nlargest(10, self._input_iterator())
with self.output().open('w') as out_file:
for streams, artist in top_10:
print(self.date_interval.date_a, self.date_interval.date_b, artist, streams, file=out_file)
def _input_iterator(self):
with self.input().open('r') as in_file:
for line in in_file:
artist, streams = line.strip().split()
yield int(streams), int(artist)
The most interesting thing here is that this task (*Top10Artists*)
defines a dependency on the previous task (*AggregateArtists*).
This means that if the output of *AggregateArtists* does not exist,
the task will run before *Top10Artists*.
.. code-block:: console
$ luigi --module examples.top_artists Top10Artists --local-scheduler --date-interval 2012-07
This will run both tasks.
Step 3 - Insert into PostgresThis mainly serves as an example of a specific subclass Task that
doesn't require any code to be written.
It's also an example of how you can define task templates that
you can reuse for a lot of different tasks.
.. code:: python
class ArtistToplistToDatabase(luigi.contrib.postgres.CopyToTable):
date_interval = luigi.DateIntervalParameter()
use_hadoop = luigi.BoolParameter()
host = "localhost"
database = "toplists"
user = "luigi"
password = "abc123" # ;)
table = "top10"
columns = [("date_from", "DATE"),
("date_to", "DATE"),
("artist", "TEXT"),
("streams", "INT")]
def requires(self):
return Top10Artists(self.date_interval, self.use_hadoop)Just like previously, this defines a recursive dependency on the
previous task. If you try to build the task, that will also trigger
building all its upstream dependencies.
Using the Central Planner
The ``--local-scheduler`` flag tells Luigi not to connect to a central scheduler.
This is recommended in order to get started and or for development purposes.
At the point where you start putting things in production
we strongly recommend running the central scheduler server.
In addition to providing locking
so that the same task is not run by multiple processes at the same time,
this server also provides a pretty nice visualization of your current work flow.
If you drop the ``--local-scheduler`` flag,
your script will try to connect to the central planner,
by default at localhost port 8082.
If you run
.. code-block:: console
$ luigid
in the background and then run your task without the ``--local-scheduler`` flag,
then your script will now schedule through a centralized server.
You need `Tornado <http://www.tornadoweb.org/>`__ for this to work.
Launching http://localhost:8082 should show something like this:
.. figure:: web_server.png
:alt: Web server screenshot
Web server screenshot
Looking at the dependency graph
for any of the tasks yields something like this:
.. figure:: aggregate_artists.png
:alt: Aggregate artists screenshot
Aggregate artists screenshot
In production, you'll want to run the centralized scheduler.
See: :doc:`central_scheduler` for more information.
---
## File: doc/execution_model.rst
Execution Model
---------------
Luigi has a quite simple model for execution and triggering.
Workers and task executionThe most important aspect is that no execution is transferred.
When you run a Luigi workflow,
the worker schedules all tasks, and
also executes the tasks within the process.
.. figure:: execution_model.png
:alt: Execution modelThe benefit of this scheme is that
it's super easy to debug since all execution takes place in the process.
It also makes deployment a non-event.
During development,
you typically run the Luigi workflow from the command line,
whereas when you deploy it,
you can trigger it using crontab or any other scheduler.
The downside is that Luigi doesn't give you scalability for free.
In practice this is not a problem until you start running thousands of tasks.
Isn't the point of Luigi to automate and schedule these workflows?
To some extent.
Luigi helps you encode the dependencies of tasks and build up chains.
Furthermore, Luigi's scheduler makes sure that there's a centralized view of the dependency graph and
that the same job will not be executed by multiple workers simultaneously.
Scheduler
A client only starts the ``run()`` method of a task when the single-threaded
central scheduler has permitted it. Since the number of tasks is usually very
small (in comparison with the petabytes of data one task is processing), we
can afford the convenience of a simple centralised server.
.. figure:: https://tarrasch.github.io/luigid-basics-jun-2015/img/50.gif
:alt: Scheduling gif
The gif is from `this presentation
<https://tarrasch.github.io/luigid-basics-jun-2015/>`__, which is about the
client and server interaction.
Triggering tasksLuigi does not include its own triggering, so you have to rely on an external scheduler
such as crontab to actually trigger the workflows.
In practice, it's not a big hurdle because Luigi avoids all the mess typically caused by it.
Scheduling a complex workflow is fairly trivial using eg. crontab.
In the future, Luigi might implement its own triggering.
The dependency on crontab (or any external triggering mechanism) is a bit awkward and it would be nice to avoid.
Trigger example
^^^^^^^^^^^^^^^
For instance, if you have an external data dump that arrives every day and that your workflow depends on it,
you write a workflow that depends on this data dump.
Crontab can then trigger this workflow every minute to check if the data has arrived.
If it has, it will run the full dependency graph.
.. code:: python
# my_tasks.py
class DataDump(luigi.ExternalTask):
date = luigi.DateParameter()
def output(self): return luigi.contrib.hdfs.HdfsTarget(self.date.strftime('/var/log/dump/%Y-%m-%d.txt'))
class AggregationTask(luigi.Task):
date = luigi.DateParameter()
window = luigi.IntParameter()
def requires(self): return [DataDump(self.date - datetime.timedelta(i)) for i in xrange(self.window)]
def run(self): run_some_cool_stuff(self.input())
def output(self): return luigi.contrib.hdfs.HdfsTarget('/aggregated-%s-%d' % (self.date, self.window))
class RunAll(luigi.Task):
''' Dummy task that triggers execution of a other tasks'''
def requires(self):
for window in [3, 7, 14]:
for d in xrange(10): # guarantee that aggregations were run for the past 10 days
yield AggregationTask(datetime.date.today() - datetime.timedelta(d), window)In your cronline you would then have something like
.. code:: console
30 0 * * * my-user luigi RunAll --module my_tasksYou can trigger this as much as you want from crontab, and
even across multiple machines, because
the central scheduler will make sure at most one of each AggregationTask task is run simultaneously.
Note that this might actually mean multiple tasks can be run because
there are instances with different parameters, and
this can give you some form of parallelization
(eg. AggregationTask(2013-01-09) might run in parallel with AggregationTask(2013-01-08)).
Of course,
some Task types (eg. HadoopJobTask) can transfer execution to other places, but
this is up to each Task to define.
File: doc/index.rst
.. Luigi documentation master file, created by
sphinx-quickstart on Sat Feb 8 00:56:43 2014.
You can adapt this file completely to your liking, but it should at least
contain the root toctree directive.
.. include:: ../README.rst
Table of Contents
.. toctree::
:maxdepth: 2
example_top_artists.rst
workflows.rst
tasks.rst
parameters.rst
running_luigi.rst
central_scheduler.rst
execution_model.rst
luigi_patterns.rst
configuration.rst
logging.rst
design_and_limitations.rst
mypy.rst
API Reference
.. autosummary::
:toctree: api
:recursive:
luigi
Indices and tables
- :ref:
genindex - :ref:
modindex - :ref:
search
File: doc/logging.rst
Configure logging
Config options:
Some config options for config [core] section
log_level
The default log level to use when no logging_conf_file is set. Must be
a valid name of a `Python log level
<https://docs.python.org/3/library/logging.html#logging-levels>`_.
Default is ``DEBUG``.
logging_conf_file
Location of the logging configuration file.
no_configure_logging
If true, logging is not configured. Defaults to false.
Config section
~~~~~~~~~~~~~~
If you're use TOML for configuration file, you can configure logging
via ``logging`` section in this file. See `example
<https://github.com/spotify/luigi/blob/master/examples/config.toml>`_
for more details.
Luigid CLI options:--background
Run daemon in background mode. Disable logging setup
and set up log level to INFO for root logger.--logdir
set logging with INFO level and output in $logdir/luigi-server.log file
Worker CLI options:
``--logging-conf-file``
Configuration file for logging.
``--log-level``
Default log level.
Available values: NOTSET, DEBUG, INFO, WARNING, ERROR, CRITICAL.
Default DEBUG. See `Python documentation
<https://docs.python.org/3/library/logging.html#logging-levels>`_
For information about levels difference.
Configuration options resolution order:- no_configure_logging option
--background--logdir--logging-conf-file- logging_conf_file option
loggingsection--log-level- log_level option
File: doc/luigi_patterns.rst
Luigi Patterns
Code Reuse
One nice thing about Luigi is that it's super easy to depend on tasks defined in other repos.
It's also trivial to have "forks" in the execution path,
where the output of one task may become the input of many other tasks.
Currently, no semantics for "intermediate" output is supported,
meaning that all output will be persisted indefinitely.
The upside of that is that if you try to run X -> Y, and Y crashes,
you can resume with the previously built X.
The downside is that you will have a lot of intermediate results on your file system.
A useful pattern is to put these files in a special directory and
have some kind of periodical garbage collection clean it up.
Triggering Many TasksA convenient pattern is to have a dummy Task at the end of several
dependency chains, so you can trigger a multitude of pipelines by
specifying just one task in command line, similarly to how e.g. make <http://www.gnu.org/software/make/>_
works.
.. code:: python
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)This simple task will not do anything itself, but will invoke a bunch of
other tasks. Per each invocation, Luigi will perform as many of the pending
jobs as possible (those which have all their dependencies present).
You'll need to use :class:~luigi.task.WrapperTask for this instead of the usual Task class, because this job will not produce any output of its own, and as such needs a way to indicate when it's complete. This class is used for tasks that only wrap other tasks and that by definition are done if all their requirements exist.
Triggering recurring tasks
A common requirement is to have a daily report (or something else)
produced every night. Sometimes for various reasons tasks will keep
crashing or lacking their required dependencies for more than a day
though, which would lead to a missing deliverable for some date. Oops.
To ensure that the above AllReports task is eventually completed for
every day (value of date parameter), one could e.g. add a loop in
requires method to yield dependencies on the past few days preceding
self.date. Then, so long as Luigi keeps being invoked, the backlog of
jobs would catch up nicely after fixing intermittent problems.
Luigi actually comes with a reusable tool for achieving this, called
:class:`~luigi.tools.range.RangeDailyBase` (resp. :class:`~luigi.tools.range.RangeHourlyBase`). Simply putting
.. code-block:: console
luigi --module all_reports RangeDailyBase --of AllReports --start 2015-01-01
in your crontab will easily keep gaps from occurring from 2015-01-01
onwards. NB - it will not always loop over everything from 2015-01-01
till current time though, but rather a maximum of 3 months ago by
default - see :class:`~luigi.tools.range.RangeDailyBase` documentation for this and more knobs
for tweaking behavior. See also Monitoring below.
Efficiently triggering recurring tasksRangeDailyBase, described above, is named like that because a more
efficient subclass exists, :class:~luigi.tools.range.RangeDaily (resp. :class:~luigi.tools.range.RangeHourly), tailored for
hundreds of task classes scheduled concurrently with contiguousness
requirements spanning years (which would incur redundant completeness
checks and scheduler overload using the naive looping approach.) Usage:
.. code-block:: console
luigi --module all_reports RangeDaily --of AllReports --start 2015-01-01It has the same knobs as RangeDailyBase, with some added requirements.
Namely the task must implement an efficient bulk_complete method, or
must be writing output to file system Target with date parameter value
consistently represented in the file path.
Backfilling tasks
Also a common use case, sometimes you have tweaked existing recurring
task code and you want to schedule recomputation of it over an interval
of dates for that or another reason. Most conveniently it is achieved
with the above described range tools, just with both start (inclusive)
and stop (exclusive) parameters specified:
.. code-block:: console
luigi --module all_reports RangeDaily --of AllReportsV2 --start 2014-10-31 --stop 2014-12-25
Propagating parameters with RangeSome tasks you want to recur may include additional parameters which need to be configured.
The Range classes provide a parameter which accepts a :class:~luigi.parameter.DictParameter
and passes any parameters onwards for this purpose.
.. code-block:: console
luigi RangeDaily --of MyTask --start 2014-10-31 --of-params '{"my_string_param": "123", "my_int_param": 123}'Alternatively, you can specify parameters at the task family level (as described :ref:here ),
however these will not appear in the task name for the upstream Range task which
can have implications in how the scheduler and visualizer handle task instances.
.. code-block:: console
luigi RangeDaily --of MyTask --start 2014-10-31 --MyTask-my-param 123.. _batch_method:
Batching multiple parameter values into a single run
Sometimes it'll be faster to run multiple jobs together as a single
batch rather than running them each individually. When this is the case,
you can mark some parameters with a batch_method in their constructor
to tell the worker how to combine multiple values. One common way to do
this is by simply running the maximum value. This is good for tasks that
overwrite older data when a newer one runs. You accomplish this by
setting the batch_method to max, like so:
.. code-block:: python
class A(luigi.Task):
date = luigi.DateParameter(batch_method=max)
What's exciting about this is that if you send multiple As to the
scheduler, it can combine them and return one. So if
``A(date=2016-07-28)``, ``A(date=2016-07-29)`` and
``A(date=2016-07-30)`` are all ready to run, you will start running
``A(date=2016-07-30)``. While this is running, the scheduler will show
``A(date=2016-07-28)``, ``A(date=2016-07-29)`` as batch running while
``A(date=2016-07-30)`` is running. When ``A(date=2016-07-30)`` is done
running and becomes FAILED or DONE, the other two tasks will be updated
to the same status.
If you want to limit how big a batch can get, simply set max_batch_size.
So if you have
.. code-block:: python
class A(luigi.Task):
date = luigi.DateParameter(batch_method=max)
max_batch_size = 10
then the scheduler will batch at most 10 jobs together. You probably do
not want to do this with the max batch method, but it can be helpful if
you use other methods. You can use any method that takes a list of
parameter values and returns a single parameter value.
If you have two max batch parameters, you'll get the max values for both
of them. If you have parameters that don't have a batch method, they'll
be aggregated separately. So if you have a class like
.. code-block:: python
class A(luigi.Task):
p1 = luigi.IntParameter(batch_method=max)
p2 = luigi.IntParameter(batch_method=max)
p3 = luigi.IntParameter()
and you create tasks ``A(p1=1, p2=2, p3=0)``, ``A(p1=2, p2=3, p3=0)``,
``A(p1=3, p2=4, p3=1)``, you'll get them batched as
``A(p1=2, p2=3, p3=0)`` and ``A(p1=3, p2=4, p3=1)``.
Note that batched tasks do not take up :ref:`resources-config`, only the
task that ends up running will use resources. The scheduler only checks
that there are sufficient resources for each task individually before
batching them all together.
Tasks that regularly overwrite the same data source
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
If you are overwriting of the same data source with every run, you'll
need to ensure that two batches can't run at the same time. You can do
this pretty easily by setting batch_method to max and setting a unique
resource:
.. code-block:: python
class A(luigi.Task):
date = luigi.DateParameter(batch_method=max)
resources = {'overwrite_resource': 1}
Now if you have multiple tasks such as ``A(date=2016-06-01)``,
``A(date=2016-06-02)``, ``A(date=2016-06-03)``, the scheduler will just
tell you to run the highest available one and mark the lower ones as
batch_running. Using a unique resource will prevent multiple tasks from
writing to the same location at the same time if a new one becomes
available while others are running.
Avoiding concurrent writes to a single file
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Updating a single file from several tasks is almost always a bad idea, and you
need to be very confident that no other good solution exists before doing this.
If, however, you have no other option, then you will probably at least need to ensure that
no two tasks try to write to the file _simultaneously_.
By turning 'resources' into a Python property, it can return a value dependent on
the task parameters or other dynamic attributes:
.. code-block:: python
class A(luigi.Task):
...
@property
def resources(self):
return { self.important_file_name: 1 }
Since, by default, resources have a usage limit of 1, no two instances of Task A
will now run if they have the same `important_file_name` property.
Decreasing resources of running tasks
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
At scheduling time, the luigi scheduler needs to be aware of the maximum
resource consumption a task might have once it runs. For some tasks, however,
it can be beneficial to decrease the amount of consumed resources between two
steps within their run method (e.g. after some heavy computation). In this
case, a different task waiting for that particular resource can already be
scheduled.
.. code-block:: python
class A(luigi.Task):
# set maximum resources a priori
resources = {"some_resource": 3}
def run(self):
# do something
...
# decrease consumption of "some_resource" by one
self.decrease_running_resources({"some_resource": 1})
# continue with reduced resources
...
Monitoring task pipelines
~~~~~~~~~~~~~~~~~~~~~~~~~
Luigi comes with some existing ways in :py:mod:`luigi.notifications` to receive
notifications whenever tasks crash. Email is the most common way.
The above mentioned range tools for recurring tasks not only implement
reliable scheduling for you, but also emit events which you can use to
set up delay monitoring. That way you can implement alerts for when
jobs are stuck for prolonged periods lacking input data or otherwise
requiring attention.
.. _AtomicWrites:
Atomic Writes Problem
~~~~~~~~~~~~~~~~~~~~~
A very common mistake done by luigi plumbers is to write data partially to the
final destination, that is, not atomically. The problem arises because
completion checks in luigi are exactly as naive as running
:meth:`luigi.target.Target.exists`. And in many cases it just means to check if
a folder exist on disk. During the time we have partially written data, a task
depending on that output would think its input is complete. This can have
devestating effects, as in `the thanksgiving bug
<http://tarrasch.github.io/luigi-budapest-bi-oct-2015/#/21>`__.
The concept can be illustrated by imagining that we deal with data stored on
local disk and by running commands:
.. code-block:: console
# This the BAD way
$ mkdir /outputs/final_output
$ big-slow-calculation > /outputs/final_output/foo.data
As stated earlier, the problem is that only partial data exists for a duration,
yet we consider the data to be :meth:`~luigi.task.Task.complete` because the
output folder already exists. Here is a robust version of this:
.. code-block:: console
# This is the good way
$ mkdir /outputs/final_output-tmp-123456
$ big-slow-calculation > /outputs/final_output-tmp-123456/foo.data
$ mv --no-target-directory --no-clobber /outputs/final_output{-tmp-123456,}
$ [[ -d /outputs/final_output-tmp-123456 ]] && rm -r /outputs/final_output-tmp-123456
Indeed, the good way is not as trivial. It involves coming up with a unique
directory name and a pretty complex ``mv`` line, the reason ``mv`` need all
those is because we don't want ``mv`` to move a directory into a potentially
existing directory. A directory could already exist in exceptional cases, for
example when central locking fails and the same task would somehow run twice at
the same time. Lastly, in the exceptional case where the file was never moved,
one might want to remove the temporary directory that never got used.
Note that this was an example where the storage was on local disk. But for
every storage (hard disk file, hdfs file, database table, etc.) this procedure
will look different. But do every luigi user need to implement that complexity?
Nope, thankfully luigi developers are aware of these and luigi comes with many
built-in solutions. In the case of you're dealing with a file system
(:class:`~luigi.target.FileSystemTarget`), you should consider using
:meth:`~luigi.target.FileSystemTarget.temporary_path`. For other targets, you
should ensure that the way you're writing your final output directory is
atomic.
Sending messages to tasks
~~~~~~~~~~~~~~~~~~~~~~~~~
The central scheduler is able to send messages to particular tasks. When a running task accepts
messages, it can access a `multiprocessing.Queue <https://docs.python.org/3/library/multiprocessing.html#pipes-and-queues>`__
object storing incoming messages. You can implement custom behavior to react and respond to
messages:
.. code-block:: python
class Example(luigi.Task):
# common task setup
...
# configure the task to accept all incoming messages
accepts_messages = True
def run(self):
# this example runs some loop and listens for the
# "terminate" message, and responds to all other messages
for _ in some_loop():
# check incomming messages
if not self.scheduler_messages.empty():
msg = self.scheduler_messages.get()
if msg.content == "terminate":
break
else:
msg.respond("unknown message")
# finalize
...
Messages can be sent right from the scheduler UI which also displays responses (if any). Note that
this feature is only available when the scheduler is configured to send messages (see the :ref:`scheduler-config` config), and the task is configured to accept them.
Gathering custom metrics from tasks' executions
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
The central scheduler is able to gather custom metrics from tasks' executions with help of
custom metrics collector (see the :ref:`scheduler-config` config). To obtain custom metrics,
you need to implement:
#. Custom metrics collector class inheriting from
:class:`~luigi.metrics.MetricsCollector` (or derived) and implementing the
:meth:`~luigi.metrics.MetricsCollector.handle_task_statistics`
method (default one does nothing). This method will be called for each task
that has been executed everytime, when
:meth:`~luigi.worker.TaskStatusReporter.report_task_statistics` is called.
For instance, following metrics collector adds monitoring tasks' execution
time and memory usage:
.. code-block:: python
class MetricsCollector(PrometheusMetricsCollector):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.task_run_execution_time = Gauge(
'luigi_task_run_execution_time_seconds',
'luigi task run method execution time in seconds',
self.labels,
registry=self.registry
)
self.task_execution_memory = Gauge(
'luigi_task_max_memory_megabytes',
'luigi task run method max memory usage in megabytes',
self.labels,
registry=self.registry
)
def handle_task_statistics(self, task, statistics):
if "elapsed" in statistics:
self.task_run_execution_time.labels(**self._generate_task_labels(task)).set(statistics["elapsed"])
if "memory" in statistics:
self.task_execution_memory.labels(**self._generate_task_labels(task)).set(statistics["memory"])
#. Custom task context manager (see the :ref:`worker-config` config),
which in `__exit__` method would call
:meth:`~luigi.worker.TaskStatusReporter.report_task_statistics` method with
the statistics dictionary. For instance, following task context manager collects
task execution time and memory usage:
.. code-block:: python
class TaskContext:
def __init__(self, task_process):
self._task_process = task_process
self._start = None
def __enter__(self):
self._start = time.perf_counter()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
assert self._start is not None
elapsed = time.perf_counter() - self._start
used_memory = max(
resource.getrusage(resource.RUSAGE_SELF).ru_maxrss, resource.getrusage(resource.RUSAGE_CHILDREN).ru_maxrss
)
logging.getLogger("luigi-interface").info(
f'Task {self._task_process.task}: time: {elapsed:.2f}s, memory: {used_memory / 1024:.2f}MB '
)
self._task_process.status_reporter.report_task_statistics({"memory": used_memory / 1024, "elapsed": elapsed})
---
## File: doc/mypy.rst
Mypy plugin
--------------
Mypy plugin provides type checking for ``luigi.Task`` using Mypy.
Require Python 3.8 or later.
How to use
~~~~~~~~~~
Configure Mypy to use this plugin by adding the following to your ``mypy.ini`` file:
.. code:: ini
[mypy]
plugins = luigi.mypy
or by adding the following to your ``pyproject.toml`` file:
.. code:: toml
[tool.mypy]
plugins = ["luigi.mypy"]
Then, run Mypy as usual.
Examples
~~~~~~~~
For example the following code linted by Mypy:
.. code:: python
import luigi
class MyTask(luigi.Task):
foo: int = luigi.IntParameter()
bar: str = luigi.Parameter()
MyTask(foo=1, bar='2') # OK
MyTask(foo='1', bar='2') # Error: Argument 1 to "Foo" has incompatible type "str"; expected "int"
## 2. Official Technical Reference & Guides (spotify/spotify.github.io)
# spotify.github.io
Showcase site for hand-picked open-source projects by Spotify. It is built using Jekyll & GitHub Actions with a Node.js script to fetch data from the GitHub GraphQL API adding it to a static YAML file in the repository.
## Development
**Install Jekyll & bundler gems**
```sh
gem install jekyll bundler
```
**Install yarn**
```sh
npm install yarn
```
**Install dependencies** inside of the project folder
```sh
yarn & bundle install
```
**Build & serve**
```sh
bundle exec jekyll serve
```
**(Optional) Test & update data**
[Create a personal GitHub Access Token](https://github.com/settings/tokens) to fetch & update the repository data locally. As the data is updated automatically using GitHub Actions `_data/projects_generated.yaml` should not be added to Git.
```sh
GH_TOKEN=YOUR_TOKEN node ./scripts/nightly.js
```
---
This project adheres to the [Open Code of Conduct][code-of-conduct]. By participating, you are expected to honor this code.
[code-of-conduct]: https://github.com/spotify/code-of-conduct/blob/master/code-of-conduct.md