1   Overview¶

This document describes the preliminary design for the SuperTask Library, an abstraction layer and set of utilities that are intended to allow LSST’s algorithmic pipeline code to be executed across a broad range of computing environments, usage contexts, and input/output storage systems. A key goal of SuperTask is to minimize any per-environment per-algorithm customization, allowing a cleaner divide between the responsibilities of algorithm developers, workflow/control system developers, and operators.

The smallest pieces of algorithmic code managed by this system are concrete SuperTasks, which inherit from the abstract base class provided by the library. The library will contain code to combine a set of SuperTasks (called a Pipeline) with a specification of which units of data to process to produce a description of the processing to be done that can be consumed by workflow systems.

Some elements of this design are still unspecified, because they rely heavily on the capabilities and interfaces of the Butler data access abstraction layer. The SuperTask Library design necessarily puts new requirements on how Butler manages and reports relationships between datasets, and a Butler redesign is currently in progress to address these (and other) requirements.

2   Task/Config Context¶

The design for the SuperTask Library sits on top of the existing LSST concepts of Task and Config classes. All SuperTasks are Tasks, and utilize the same Config system for algorithmic configuration parameters.

A concrete Task is simply a configurable, composable, callable object. While all Tasks inherit from a common abstract base class and define a run method, each Task defines its own signature for run, so Tasks do not really share a common interface. The lsst.pipe.base.Task abstract base class itself exists largely to provide utility code to its subclasses, such as setting up logging, providing objects to hold processing metadata, and setting up nested Tasks to be delegated to, called subtasks.

This composition of Tasks is closely tied to our approach for configuring them, and it is this functionality that makes the Task concept so useful. Configuration options for a Task are defined by a corresponding class that inherits from lsst.pex.config.Config. Config classes contain lsst.pex.config.Field instances, which act like introspectable properties that define the types, allowed values, and documentation for configuration options. A set of configuration values for a Task is thus just an instance of its Config class, and overrides for those values can be expressed as Python code that assigns values to to the Field attributes. When a Task delegates to another as a subtask, its Config class usually contains a special ConfigurableField that holds an instance of the subtask’s Config class. ConfigurableField allows this Config instance to be replaced by one for a different Task, allowing the subtask to be replaced by another with the same run signature.

The SuperTask abstract base class inherits from Task, and its concrete subclasses are expected to defined a Config class to define their configuration parameters and delegate additional work to subtasks. Using a SuperTask as a subtask is not meaningful, however; in that context the SuperTask just behaves like a regular Task and the additional interfaces and functionality added by SuperTask go unused (as a result, we expect this to be rare).

A few additional properties of Tasks are particularly relevant for SuperTask design:

• The configuration of a Task is frozen after the Task is constructed.
• The schema of any catalogs produced by a Task must be fully defined after Task construction, and must not depend on the actual contents of any data products.
• Calls to run or any other methods must not change any internal state.

3   Functional Design and Usage Pattern¶

The design of the SuperTask Library is largely derived from the following two principles:

• Defining units of work that can be performed independently should be a responsibility of the same class (a concrete SuperTask, in this case) that does that work.

Putting this responsibility on the control software or the human user instead would result in a rigid system that is capable of running only a few predefined sequences of SuperTask without requiring significant changes. While we will likely only need a few predefined sequences late in operations, we need more flexibility during development and early operations.

• By requesting a list of these units of work from each SuperTask in an ordered list, the control software can discover all dependencies and construct a satisfactory execution plan, in advance, for the full sequence of SuperTasks.

This does not allow the definition of a particular SuperTask’s units of work to depend on the actual outputs of an earlier SuperTask in the sequence (as opposed to depending on just the expected presence or absence of outputs, which is supported).

We consider this limitation acceptable for two reasons. First, we expect cases where the outputs themselves affect the dependencies to be rare, and hence it is an acceptable fallback to simply split the list of SuperTasks into subsets without these dependencies and run the subsets in sequence manually, because the number of such subsets will be small. More importantly, we believe we can strictly but only slightly overestimate the dependencies between units of work in advance, in essentially all of these cases, and hence the only errors in the execution plan will be a small number of do-nothing jobs and/or unnecessary inputs staged to the local compute environment. These can easily be handled by any practical workflow system.

For the remainder of this document, we will refer to an independent unit of work performed by a SuperTask (and the list of input and output datasets involved) as a Quantum. An ordered list of SuperTasks (which includes their configuration) is what we call a Pipeline. The control software has many components with different responsibilities, which we will introduce in the remainder of this section.

The typical usage pattern for the SuperTask Library is as follows.

#. A developer defines a Pipeline from a sequence of SuperTasks, including their configuration, either programmatically or by editing a TBD text-based, human-readable file format. Other developers may then modify the Pipeline to modify configuration or insert or delete SuperTasks, again via either approach.

#. An operator passes the Pipeline, an input data repository, and a Data ID Expression (see 6   Relating and Specifying Data IDs) to a PreFlightFramework. Different PreFlightFrameworks will be implemented for different contexts. Some PreFlightFrameworks may provide an interface for making a final round of modifications to the Pipeline at this stage, but these modifications are not qualitatively different from those in the previous step.

1. The PreFlightFramework passes the Pipeline, the input data repository, and the Data ID Expression to a GraphBuilder (see 7   Pre-Flight Environment), which
   • inspects the Pipeline to construct a list of all dataset types consumed and/or produced by the Pipeline;
   • queries the data repository to obtain a RepoGraph that contains all datasets of these types that match the given Data ID Expression (see 6   Relating and Specifying Data IDs);
   • calls the defineQuanta method of each SuperTask in the Pipeline in sequence, accumulating a list of all quanta to be executed;
   • constructs the Quantum Graph (see 7   Pre-Flight Environment), a bipartate directed acyclic graph with quantum vertices linked by the dataset vertices they produce and consume.

#. The Quantum Graph is passed to an ExecutionFramework, along with additional configuration for how the processing is to be performed (changes in this configuration must not change the outputs of the Pipeline except to allow intermediate datasets to be elided). The ExecutionFramework may be the same class as the PreFlightFramework (as in lsst.pipe.base.CmdLineTask, which performs both roles), which makes this step a no-op. It may also be a completely different class that may be run in an entirely different compute environment (via a serialized Quantum Graph).

1. The ExecutionFramework creates one or more output data repositories and records in them any repository-wide provenance (such as the Pipeline configuration or software versions).

#. The ExecutionFramework walks the Quantum Graph according to the partial ordering it defines, and calls runQuantum on the appropriate concrete SuperTask for each quantum vertex. Depending on the activator, the SuperTasks may be run directly in the same compute environment, or submitted to a workflow system for execution elsewhere (probably by translating the generic Quantum Graph to a format specific to a particular workflow system). In some environments a temporary local data repository containing only the datasets consumed by a particular set of quanta may be created in scratch space to support execution in a context in which the original data repositories are not accessible, with output datasets similarly staged back to the true output data repositories.

4   SuperTask Class Interface¶

class SuperTask(Task)¶

__init__(self, butler=None, **kwargs)¶

All concrete SuperTasks must have the __init__() signature shown here, in which **kwargs contains only arguments to be forwarded to Task.__init__ (additional keyword-only arguments are also allowed, as long as they have default values). The abstract base class does not use the butler argument, allowing it to be None, and while concrete SuperTasks may or may not use it, they must accept it even if it is unused. This allows the schemas associated with input dataset types and the configuration of preceeding SuperTasks to be loaded and used to complete construction of the SuperTask; a SuperTask should not assume any other datasets are available through the given Butler. SuperTasks that do use the butler argument should also provide an alternate way to provide the schemas and configuration (i.e. additional defaulted keyword arguments) to allow them to be constructed without a Butler when used as a regular Task. This also implies that when a Pipeline constructs a sequence of SuperTasks, it must ensure the schemas and configuration are recorded at each step, not just at the end.

run(self, *args, **kwargs)¶

This is the standard entry point for all Tasks, with the signature completely different for each concrete Task.

This should perform the bulk of the SuperTask’s algorithmic work, operating on in-memory objects for both arguments and return values, and should not utilize a Butler or perform any I/O. In rare cases, a SuperTask for which I/O is an integral component of the algorithm may lack a run() method, or may have multiple methods to serve the same purpose. As with other Tasks, the return value should be a lsst.pipe.base.Struct combining named result objects.

defineQuanta(self, repoGraph)¶

Called during pre-flight, in this method a concrete SuperTask subdivides work into independently-executable units (quanta) and relates the input datasets of these to their output datasets. The only argument is a RepoGraph instance, a graph object describing the current state of the relevant subset of the input data repository. On return, the RepoGraph should be modified to additionally contain datasets that will be produced by the SuperTask, reflecting the fact that they will be present in the data repository by the time subsequent SuperTask’s in the same Pipeline are executed. The return value should be a list of Quantum instances.

runQuantum(self, quantum, butler)¶

This method runs the SuperTask on the given Quantum, using a Butler for input and output. For most concrete SuperTasks, this should simply use Butler.get to retrieve inputs, call run, and then use Butler.put to write outputs.

getDatasetClasses(self)¶

Called during pre-flight (before defineQuanta()), this method returns the sets of input and output Datasets classes used by this SuperTask. As long as DatasetField is used to control the Dataset classes utilized by the SuperTask's, the default implementation provided by the SuperTask base class itself should be sufficient.

getDatasetSchemas(self)¶

This method returns a dict containing the schemas that correspond to any table-like datasets output by the SuperTask. Dictionary keys are Dataset types. This may be extended in the future to contain other schema-like information for non-table datasets.

5   Pipeline Class Interface¶

class Pipeline¶

Pipeline behaves like (and should probably be implemented as) a thin layer over Python’s built-in OrderedDict, in which the dictionary values hold a concrete SuperTask subclass and its configuration and the keys are simply string labels. The order of the items must be consistent with the partial ordering implied by the sequence of Dataset classes used by the concrete SuperTasks, though this is condition is only checked on request – trying to maintain it as a class invariant would make it much more difficult to modify the Pipeline in-place.

checkOrder(self)¶

Return False if any SuperTask in the py:class:Pipeline produces an output Dataset that has already been utilized as an input by a SuperTask that appears earlier in the Pipeline's iteration order.

sort(self):

Modify the iteration order of the Pipeline to guarantee that subsequent calls to checkOrder() will return True.

applyConfigOverrides(self, overrides)¶

Apply a set of configuration overrides to the SuperTask labeled with the given key. The overrides are given as a dictionary with keys matching labels for SuperTasks in the Pipeline, and values holding configuration overrides for that SuperTask.

Note

This assumes a Python class representing a set of config overrides, which pex_config currently does not provide.

6   Relating and Specifying Data IDs¶

7   Pre-Flight Environment¶

With the class interfaces described in the last few sections, we can now more fully describe the “pre-flight” procedure summarized in Section 3   Functional Design and Usage Pattern. Unlike the quantum execution environment, most of preflight is common code shared by all PreFlightFrameworks, which simply provide different front-end APIs appropriate for their users and supply an appropriate implementation of makeRepoGraph() for the given input data repository.

The inputs to all PreFlightFrameworks (though one or more may be defaulted) are:

• The input data repository or a Butler initialized to point to it.
• A user expression defining the units of data to process, in the form of a SQL WHERE clause that can be passed directly to makeRepoGraph().
• A Pipeline instance.

A PreFlightFramework delegates essentially all remaining work to the QuantumGraphBuilder:

• The PreFlightFramework constructs a QuantumGraphBuilder, passing it the Pipeline instance.

The QuantumGraphBuilder instantiates all SuperTasks in the Pipeline, collecting their (now frozen) configuration, schemas, and input and output dataset types.

• The PreFlightFramework creates a RepoGraph from the input data repository, the user WHERE expression, and the lists of dataset types reported by the QuantumGraphBuilder by calling py:func:`makeRepoGraph.

The design also leaves open the possibility that the operations PreFlightFramework will construct a RepoGraph by some other means (which could support a more complicated set of SQL queries that target an operations-specific SQL schema).

• QuantumGraphBuilder.makeGraph() is called with the RepoGraph to build the Quantum Graph.

A more detailed description of QuantumGraphBuilder is below.

class QuantumGraphBuilder¶

__init__(self, pipeline, butler)¶

The :py:class:GraphBuilder` first iterates over the SuperTasks in the Pipeline, instantiating them (which freezes their configuration), and accumulating a list of input and output dataset types by calling SuperTask.getDatasetClasses() on each. Dictionaries containing configuration and schemas are also constructed for later use in recording provenance.

Note

While instantiating a SuperTask in general requires a Butler, this is mostly to allow downstream SuperTasks to obtain the schemas of their input dataset types. While there’s no way to avoid having QuantumGraphBuilder use the given Butler to load the schemas of the overall input dataset types (assuming any of these are catalogs), it could use a dummy Butler backed by a simple dict to transfer schemas obtained by calling SuperTask.getDatasetSchemas() to downstream SuperTask.__init__(). At the same time, it would build up its own py:attr:schemas attribute, which could be used by the PreFlightFramework to actually persist the schemas.

NeededDatasets¶

A set of dataset types (subclasses of Dataset) that are used strictly as inputs by the Pipeline the QuantumGraphBuilder was constructed with.

FutureDatasets¶

A set of dataset types (subclasses of Dataset) that are produced as outputs (including intermediates) by the Pipeline the QuantumGraphBuilder was constructed with.

configs¶

A dict mapping SuperTask name to Config instance.

makeGraph(self, repoGraph)¶

Construct a QuantumGraph representing (conceptually) the processing to be performed and its dependencies.

This is implemented by iterating through the SuperTasks instantiated by __init__(), calling SuperTask.defineQuanta() with the RepoGraph.

As each SuperTask defines its quanta, it also adds the Datasets it will produce to the RepoGraph, making it appear to subsequent SuperTasks that these datasets are already present in the repository and may be used as inputs. The result of this iteration is a sequence of Quantum instances.

The final step is to transfrom this sequence into the Quantum Graph, which is a directed acyclic graph describing the dependencies of the processing. Each node in the Quantum Graph is conceptually either a Quantum or a Dataset, with the direction of the graph edges representing inputs (Dataset node to Quantum node) and outputs (Quantum node to Dataset node). Because each Quantum instance holds its input and output Dataset instances, the only remaining step to making the sequence of quanta into a fully-walkable graph is to add back-references from each Dataset, filling in its creator and consumers attributes to point to the appropriate Quantum instances.

class QuantumGraph¶

The attributes that connect Quanta to Datasets naturally form a graph data structure, which we call a QuantumGraph.

Because the graph structure is mostly defined by its constituent classes, QuantumGraph simply provides flat access to these.

quanta¶

A list of Quantum instances, ordered in a way that satisfies dependencies (which may be unique).

datasets¶

A dictionary with dataset type names as keys and sets of Dataset instances of that type as values.

8   Quantum-Execution Environment¶

Unlike the pre-flight environment, the code that implements the quantum execution environment in which SuperTask.runQuantum() is called and actual algorithmic code is run is in general not shared between different implementations.

A QuantumExecutionFramework can be as simple as a thin layer that provides a call to SuperTask.runQuantum() with a Butler or as complex as a multi-level workflow system that involves staging data to local filesystems, strict provenance control, multiple batch submissions, and automatic retries.

At the lowest level, all QuantumExecutionFrameworks will have to do at least the following tasks:

• Instantiate one or more SuperTasks from the Pipeline (ensuring that this is done consistently with how they were instantiated in pre-flight). This also involves initializing logging for SuperTask(s) and their subtasks, and will require setting up a Butler (possibly a simple dict-backed one) to facilitate the transfer of schema information.
• Create a Butler (possibly the same as the one used for SuperTask construction).
• Call SuperTask.runQuantum() on each of the Quantum instances it is responsible for running.

When careful control over provenance is necessary, the Butler passed to SuperTask.runQuantum() can be instrumented to detect the actual datasets loaded by the task, though even this probably cannot fully replace reporting by the task itself about what was used.

When data is staged to a local filesystem for execution, the Butler created in the local filesystem only needs to provide the capability to get and put the input and output datasets that are included in the quanta to be executed. Because the mappings between the Datasets in the quanta and the staged files can be fully determined at pre-flight, the Butler implementation here can be quite simple as long as the staging system can transfer an additional file containing those mappings.

9   Notes on specific expected implementations¶