The pipeline mechanism allows to apply a sequence of processing steps to some data, in a way that the input of each process is the output of the previous one. Making visible these different processing steps, in the right order, is essential in data analysis to keep track of what you did, and make sure that the whole processing remains consistent.
Pipelet is based on the possibility to save on disk every intermediate input or output of a pipeline, which is usually not a strong constraint but offers a lot of benefits. It means that you can stop the processing whenever you want, and start it again without recomputing the whole thing: you just take the last products you have on disk, and continue the processing where it stopped. This logic is interesting when the computation cost is higher than the cost of disk space required by intermediate products.
In addition, the Pipelet engine has been designed to process data sets. It takes advantage of the parallelisation opportunity that comes with data which share the same structure (data arrays), to dispatch the computational tasks on parallel architecture. The data dependency scheme is also used to save CPU time, and allows to handle very big data sets processing.
Pipelet is a free framework which helps you :
- to write and manipulate pipelines with any dependency scheme,
- to keep track of what processing has been applied to your data and perform comparisons,
- to carry pipelines source code from development to production and adapt to different hardware and software architectures.
- Speed improvement during execution and navigation to handle pipeline of 100 thousand tasks.
- Task repository versionning to manage group_by directive which uses different parent tasks list.
- New glob_seg type utility to search data files from parent task only + improvement of I/O and parameters utilities. See The segment environnement section
- Improvement of external dependencies management : the depend directive induces a copy of external dependencies, the version number (together with RCS revision if exist) of the imported modules are output.
- Pickle file render available from the Web interface
- Running the pipelet engine requires Python >= 2.6.
- The web interface of pipelet requires the installation of the cherrypy3 Python module (on Debian: aptitude install python-cherrypy3).
- Although default Python installation provides the sqlite3 module, you may not be able to use it. In that case, you can manually install the pysqlite2 module.
You may find useful to install some generic scientific tools that nicely interact with pipelet:
- numpy
- matplotlib
- latex
Note
The first version of the software is currently in the process of stabilisation. The Pipelet engine has now reached the level of desired sophistication. On the other hand, the user interface has been developped in a minimalist way. It includes the main functionalities but with a design which could and will be more user friendly.
git clone https://gitlab.in2p3.fr/pipelet/pipelet.git
sudo python setup.py install
Run the test pipeline:
cd test/first_test
python main.py
Add this pipeline to the web interface:
pipeweb track test ./.sqlstatus
Set up an account in the access control list and launch the web server:
pipeutils -a username -l 2 .sqlstatus
pipeweb start
You should be able to browse the result on the web page http://localhost:8080
To get a new pipeline framework, with example main and segment scripts:
pipeutils -c pipename
This command ends up with the creation of directory named pipename wich contains:
- a main script (named main.py) providing functionnalities to execute your pipeline in various modes (debug, parallel, batch mode, ...)
- an example of segment script (default.py) which illustrates the pipelet utilities with comments.
The next section describes those two files in more details.
The dependencies between segments must form a directed acyclic graph. This graph is described by a char string using a subset of the graphviz dot language (http://www.graphviz.org). For example the string:
"""
a -> b -> d;
c -> d;
c -> e;
"""
defines a pipeline with 5 segments {"a", "b", "c", "d", "e"}. The relation a->b ensures that the processing of the segment “a” will be done before the processing of its ‘child’ segment b. Also the output of a will be fed as input for b. In the given example, the node d has two parents b and c. Both will be executed before d. As their is no relation between b and c which of the two will be executed first is not defined.
When executing the segment seg, the engine looks for a python script named seg.py. If not found, it looks iteratively for script files named se.py and s.py. This way, different segments of the pipeline can share the same code, if they are given a name with a common root (this mechanism is useful to write generic segment and is completed by the hooking system, described in the advanced usage section). The code is then executed in a specific namespace (see below The segment environment section).
Practically, the creation of a Pipeline object requires 3 arguments:
from pipelet.pipeline import Pipeline
P = Pipeline(pipedot, code_dir="./", prefix="./")
It is possible to output the graphviz representation of the pipeline (requires the installation of graphviz). First, save the graph string into a .dot file with the pipelet function:
P.to_dot('pipeline.dot')
Then, convert it to an image file with the dot command:
dot -Tpng -o pipeline.png pipeline.dot
The modification of the code of one segment will trigger its recalculation and the recalculation of all the segments which depend on it.
The output of a segment is a list of python objects. If a segment as no particular output this list can be empty and do not need to be specified. Elements of the list are allowed to be any kind of pickleable python objects. However, a good practice is to fill the list with the minimal set of characteristics relevant to describe the output of the segment and to defer the storage of the data to appropriate structures and file formats. For example, a segment which performs computation on large images could virtually pass the results of its computation to the following segment using the output list. It is a better practice to store the resulting image in a dedicated file and to pass in the list only the information allowing a non ambiguous identification of this file (like its name or part of it).
The input of a child segment is taken in a set build from the output lists of its parents. The content of the input set is actually tunable using the multiplex directive (see below). However the simplest and default behavior of the engine is to form the Cartesian product of the output list of its parent.
To illustrate this behavior let us consider the following pipeline, build from three segments:
"""
knights -> melt;
quality -> melt;
"""
and assume that the respective output lists of segments knights and quality are:
["Lancelot", "Galahad"]
and:
['the Brave', 'the Pure']
The Cartesian product of the previous set is:
[('Lancelot','the Brave'), ('Lancelot','the Pure'), ('Galahad','the Brave'), ('Galahad','the Pure')]
Four instances of segment melt will thus be run, each one receiving as input one of the four 2-tuples.
At the end of the execution of all the instances of a segment, their output lists are concatenated. If the action of segment melt is to concatenate the two strings he receives separated by a space, the final output set of segment melt will be:
[('Lancelot the Brave'), ('Lancelot the Pure'), ('Galahad the Brave'), ('Galahad the Pure')].
This default behavior can be altered by specifying a #multiplex directive in the commentary of the segment code. See section Multiplex directive for more details.
As the segment execution order is not uniquely determined by the pipe scheme (several path may exists), it is not possible to retrieve an ordered input tuple. To overcome this issue, segment inputs are dictionaries, with keywords matching parent segment names. In the above example, one can read melt inputs using:
k = seg_input["knights"]
q = seg_input["quality"]
One can also use dedicated segment routines:
k = get_input("knights")
q = get_input("quality")
See section The segment environment for more details.
By default, orphan segments (segments without parents) have no input argument (an empty list), and therefore are executed once without input. The possibility is offered to feed input to an orphan segment by pushing a list into the output set of an hypothetic (‘phantom’) parent. If P is an instance of the pipeline object, this is done by:
P.push (segname=[1,2,3])
From the segment environment, inputs can be retrieve from the dedicated routine:
id = get_input()
In this scheme, it is important to uniquely identify the child tasks of the orphan segment by setting a dedicated output.:
seg_output = id
See section The segment environment for more details.
The framework provides versioning of your data and easy access through the web interface. It also keep track of the code, of the execution logs, and various meta-data of the processing. Of course, you remain able to bypass the hierarchical storage and store your actual data elsewhere, but you will loose the benefit of automated versioning which proves to be quite convenient.
The storage is organized as follows:
all pipeline instances are stored below a root which corresponds to the prefix parameter of the Pipeline object.
/prefix/
all segment meta data are stored below a root which name corresponds to a unique hash computed on the segment code and its dependencies.
/prefix/segname_YFLJ65/
Segment’s meta data are: - a copy of the segment python script - a copy of all segment hook scripts - a parameter file (.args) which contains segment parameters value - a meta data file (.meta) which contains some extra meta data
all segment instances data and meta data are stored in a specific subdirectory which name corresponds to a string representation of its input prefix by its identifier number
/prefix/segname_YFLJ65/data/1_a/
/prefix/segname_YFLJ65/data/
If a segment has at least one parent, its root will be located below one of its parent’s one :
/prefix/segname_YFLJ65/segname2_PLMBH9/
etc...
While the hierarchical storage makes easy the storing and handling of many different data with different versions, it can make the manual navigation in the data less convenient. Here comes the role of the [[*Browsing%20Pipes][web interface]] (among other advantages, like distant access to the data, tagging...).
The segment code is executed in a specific environment that provides:
access to the segment input and output - get_input(seg): return the input coming from segment seg. If no
segment specified, take the first. This utility replaces the seg_input variable which type could vary as described below.
Functionalities to use the automated hierarchical data storage system. - get_data_fn(basename, seg): complete the filename with the path to
the working directory of the segment (default is the current segment).
the pattern y in the data directory of direct parent tasks. It is possible to search only in a specific segment list segs.
pattern y in the data directory of parent segment x (all task directories are searched, independantly of whether the file comes from a task related to the current task). Its usage should be limited as it: - potentially breaks the dependancy scheme. - may hurt performances as all task directories of the segment
x will be searched.
get_tmp_fn(): return a temporary filename.
Functionalities to use the automated parameters handling - save_param(lst): the listed parameters will be saved in a dedicated file. - expose(lst): the listed parameters will be exposed from the web interface - load_param(seg, globals(), lst): retrieve parameters from the meta data.
Various convenient functionalities - save_products(filename, globals(), lst_par): use pickle to save a
part of the current namespace.
Hooking support Pipelet enables you to write reusable generic segments by providing a hooking system via the hook function. hook(hookname, globals()): execute Python script segname_hookname.py and update the namespace. See the section Hooking system for more details.
See the API documentation environment to get more details on those functionalities
We are now in the position to write a complete simple pipeline. Let us consider the knights example and write the beginning of the main file main.py describing the pipeline:
from pipelet.pipeline import Pipeline
pipedot `` """
knights -> melt;
quality -> melt;
"""
P = Pipeline(pipedot, code_dir='./',prefix='./')
Now, we create the 3 segment files knights.py, quality.py and melt.py. The only action we expect from segment knights is simply to provide a list of knights. Its code is very simple:
set_output(["Lancelot", "Galahad")
Same thing for the segment quality:
set_output (['the Brave', 'the Pure'] )
As explained, the segment melt will be executed four times. We expect from it to concatenate its input and write the result into a file, so the code is:
knight, quality = get_input('knights'), get_input('quality')
f = open(get_data_fn('result.txt'), 'w')
f.write(knight + ' ' + quality+'\n')
f.close()
We need to complete the main file so that it takes care of the execution (see Running pipes for more explanations):
from pipelet.pipeline import Pipeline
from pipelet.launchers import launch_interactive
pipedot = """
knights -> melt;
quality -> melt;
P = Pipeline(pipedot, code_dir='./',prefix='./')
w,t = launch_interactive(P)
w.run()
The execution of the main file will run this simple example in the ‘interactive’ mode provided for debugging purposes. You may add a knight in the list to see only the required part recomputed. More complete examples are described in the Example pipes section. The two remaining sections of the tutorial explain how to use execution mode that enable to exploitation of data parallelism (in this case running the four independent instances of the melt segment in parallel), and how to provide web access to the results.
A sample main file is made available when creating a new Pipelet framework. It is copied from the reference file
pipelet/pipelet/static/main.py
This script illustrates various ways of running pipes. It describes the different parameters, and also, how to write a main python script can be used as any binary from the command line (including options parsing).
Some options are common to each running modes.
The logging system is handle by the python logging facility module. This module defines the following log levels :
All logging messages are saved in the different Pipelet log files, available from the web interface (rotating file logging). It is also possible to print those messages on the standard output (stream logging), by setting the desired log level in the launchers options: For example:
import logging
launch_process(P, N,log_level=logging.DEBUG)
If set to 0, stream logging will be disable.
Note
The matplotlib documentation says: “Many users report initial problems trying to use maptlotlib in web application servers, because by default matplotlib ships configured to work with a graphical user interface which may require an X11 connection. Since many barebones application servers do not have X11 enabled, you may get errors if you don’t configure matplotlib for use in these environments. Most importantly, you need to decide what kinds of images you want to generate (PNG, PDF, SVG) and configure the appropriate default backend. For 99% of users, this will be the Agg backend, which uses the C++ antigrain rendering engine to make nice PNGs. The Agg backend is also configured to recognize requests to generate other output formats (PDF, PS, EPS, SVG). “
The easiest way to configure matplotlib to use Agg is to call:
matplotlib.use('Agg')
The matplotlib and matplotlib_interactive options turn the matplotlib backend to Agg in order to allow the execution in non-interactive environment. The two options affects independently the non interactive execution mode and the interactive mode.
Those two options are set to True by default in the sample main script. Setting them to False deactivate this behavior for pipelines that make no use of matplotlib (and prevents the raise of an exception if matplotlib is not even available).
This mode has been designed to ease debugging. If P is an instance of the pipeline object, the syntax reads
from pipelet.launchers import launch_interactive
w, t = launch_interactive(P)
w.run()
In this mode, each tasks will be computed in a sequential way. Do not hesitate to invoque the Python debugger from IPython :%pdb
To use the interactive mode, run:
main.py -d
In this mode, one can run simultaneous tasks (if the pipe scheme allows to do so). The number of subprocess is set by the N parameter
from pipelet.launchers import launch_process
launch_process(P, N)
To use the process mode, run:
main.py
or:
main.py -p 4
In this mode, one can submit some batch jobs to execute the tasks. The number of job is set by the N parameter
from pipelet.launchers import launch_pbs
launch_pbs(P, N , address=(os.environ['HOST'],50000))
It is possible to specify some job submission options like:
The server option can be disable to add some workers to an existing scheduler.
To use the batch mode, run:
main.py -b
to start the server, and:
main.py -a 4
to add 4 workers.
The pipelet webserver allows the browsing of multiple pipelines. Each pipeline has to be register using
pipeweb track <shortname> sqlfile
To remove a pipeline from the tracked list, use
pipeweb untrack <shortname>
As the pipeline browsing implies a disk parsing, some basic security has to be set also. All users have to be register with a specific access level (1 for read-only access, and 2 for write access).
pipeutils -a <username> -l 2 sqlfile
To remove a user from the user list:
pipeutils -d <username> sqlfile
Start the web server using
pipeweb start
Then the web application will be available on the web page http://localhost:8080
To stop the web server
pipeweb stop
In order to ease the comparison of different processing, the web interface displays various views of the pipeline data :
The index page displays a tree view of all pipeline instances. Each segment may be expand or reduce via the +/- buttons.
The parameters used in each segments are resumed and displayed with the date of execution and the number of related tasks order by status.
A check-box allows to performed operation on multiple segments :
- deletion : to clean unwanted data
- tag : to tag remarkable data
The filter panel allows to display the segment instances with respect to 2 criterions :
- tag
- date of execution
Each segment names is a link to its code page. From this page the user can view all python scripts code which have been applied to the data.
The tree view is reduced to the current segment and its related parents.
The root path corresponding to the data storage is also displayed.
The number of related tasks, order by status, is a link to the product pages, where the data can be directly displayed (if images, or text files) or downloaded. From this page it is also possible to delete a specific product and its dependencies.
The log page can be acceded via the log button of the filter panel. Logs are ordered by date.
This example illustrates a very simple image processing use. The problematic is the following : one wants to apply a Gaussian filter in Fourier domain on several 2D images.
The pipe scheme is:
pipedot = """
mkgauss->convol;
fftimg->convol;
"""
where segment mkgauss computes the Gaussian filter, fftimg computes the Fourier transforms of the input images, and convol performs the filtering in Fourier domain, and the inverse transform of the filtered images.
P = pipeline.Pipeline(pipedot, code_dir=op.abspath('./'), prefix=op.abspath('./'))
P.to_dot('pipeline.dot')
The pipe scheme is output as a .dot file, that can be converted to an image file with the command line:
dot -Tpng -o pipeline.png pipeline.dot
To apply this filter to several images (in our case 4 input images), the pipe data parallelism is used. From the main script, a 4-element list is pushed to the fftimg segment.
P.push(fftimg=[1,2,3,4])
At execution, 4 instances of the =fftimg= segment will be created, and each of them outputs one element of this list
img = get_input() #(fftimg.py - line 15)
set_output (img) #(fftimg.py - line 38)
On the other side, a single instance of the mkgauss segment will be executed, as there is one filter to apply.
The last segment convol, which depends on the two others, will be executed with a number of instances that is the Cartesian product of its 4+1 inputs (ie 4 instances)
The instance identifier which is set by the fftimg output, can be retrieve with the following instruction:
img = get_input('fftimg') #(convol.py - line 12)
Follow the same procedure than for the first example pipeline, to run this pipe and browse the result.
This CMB pipeline depends on two external python modules:
- healpy : http://code.google.com/p/healpy/
- spherelib: http://gitorious.org/spherelib
This example illustrates a very simple CMB data processing use.
The problematic is the following : one wants to characterize the inverse noise weighting spectral estimator (as applied to the WMAP 1yr data). A first demo pipeline is built to check that the algorithm has correctly been implemented. Then, Monte Carlo simulations are used to compute error bars estimates.
The design pipe scheme is:
pipe_dot = """
cmb->clcmb->clplot;
noise->clcmb;
noise->clnoise->clplot;
"""
where:
As the two first orphan segments depends on a single shared parameter which is the map resolution nside, this argument is pushed from the main script.
Another input argument of the cmb segment, is its simulation identifier, which will be used for latter Monte Carlo. In order to push two inputs to a single segment instance, we use python tuple data type.:
P.push(cmb=[(nside, 1)])
P.push(noise=[nside])
From the segment, those inputs are retrieved with :
(nside,sim_id) = get_input() ##(cmb.py line 14)
nside = seg_input() ##(noise.py line 15)
The last segment produces a plot in which we compare:
- the input LCDM power spectrum
- the binned cross spectrum of the noisy CMB maps
- the binned cross spectrum of which we applied hitcount weight and mode coupling matrix.
- the noise power spectrum computed by clnoise segment.
In this plot we check that both estimators are corrects, and that the noise level is the expected one.
As a second step, Monte Carlo simulations are used to compute error bars.
The clnoise segment is no longer needed, so that the new pipe scheme reads :
pipe_dot = """
cmb->clcmb->clplot;
noise->clcmb;
"""
We now use the native data parallelization scheme of the pipe to build many instances of the cmb and clcmb segments.
cmbin = []
for sim_id in [1,2,3,4,5,6]:
cmbin.append((nside, sim_id))
P.push(cmb=cmbin)