.. include:: ../../global.inc .. include:: manual_chapter_numbers.inc .. index:: pair: transforming in parallel; Tutorial .. _new_manual.transform_in_parallel: ###################################################################################################### |new_manual.transform_in_parallel.chapter_num|: More on ``@transform``-ing data ###################################################################################################### .. seealso:: * :ref:`Manual Table of Contents ` * :ref:`@transform ` syntax .. note:: Remember to look at the example code: * :ref:`new_manual.transform_in_parallel.code` *************************************** Review *************************************** .. image:: ../../images/theoretical_pipeline_schematic.png :scale: 50 Computational pipelines transform your data in stages until the final result is produced. *Ruffus* automates the plumbing in your pipeline. You supply the python functions which perform the data transformation, and tell *Ruffus* how these pipeline stages or :term:`task` functions are connected together. .. note:: **The best way to design a pipeline is to:** * **write down the file names of the data as it flows across your pipeline** * **write down the names of functions which transforms the data at each stage of the pipeline.** :ref:`new_manual.introduction` described the bare bones of a simple *Ruffus* pipeline. Using the *Ruffus* :ref:`@transform ` decorator, we were able to specify the data files moving through our pipeline so that our specified task functions could be invoked. This may seem like a lot of effort and complication for something so simple: a couple of simple python function calls we could have invoked ourselves. However, By letting *Ruffus* manage your pipeline parameters, you will get the following features for free: #. Only out-of-date parts of the pipeline will be re-run #. Multiple jobs can be run in parallel (on different processors if possible) #. Pipeline stages can be chained together automatically. This means you can apply your pipeline just as easily to 1000 files as to 3. ************************************ Running pipelines in parallel ************************************ Even though three sets of files have been specified for our initial pipeline, and they can be processed completely independently, by default *Ruffus* runs each of them serially in succession. To ask *Ruffus* to run them in parallel, all you have to do is to add a ``multiprocess`` parameter to ``pipeline_run``: :: >>> pipeline_run(multiprocess = 5) In this case, we are telling *Ruffus* to run a maximum of 5 jobs at the same time. Since we only have three sets of data, that is as much parallelism as we are going to get... .. _new_manual.only_rerun_out_of_date: ************************************************** Up-to-date jobs are not re-run unnecessarily ************************************************** A job will be run only if the output file timestamps are out of date. If you ran our example code a second time, nothing would happen because all the work is already complete. We can check the details by asking *Ruffus* for more ``verbose`` output :: >>> pipeline_run(verbose = 4) Task = map_dna_sequence All jobs up to date Task = compress_sam_file All jobs up to date Task = summarise_bam_file All jobs up to date Nothing happens because: * ``a.sam`` was created later than ``a.1.fastq`` and ``a.2.fastq``, and * ``a.bam`` was created later than ``a.sam`` and * ``a.statistics`` was created later than ``a.bam``. and so on... Let us see what happens if we recreated the file ``a.1.fastq`` so that it appears as if 1 out of the original data files is out of date :: open("a.1.fastq", "w") pipeline_run(multiprocess = 5) The up to date jobs are cleverly ignored and only the out of date files are reprocessed. .. code-block:: pycon :emphasize-lines: 3,4,7,8,11,12 >>> open("a.1.fastq", "w") >>> pipeline_run(verbose=2) Job = [[b.1.fastq, b.2.fastq] -> b.sam] # unnecessary: already up to date Job = [[c.1.fastq, c.2.fastq] -> c.sam] # unnecessary: already up to date Job = [[a.1.fastq, a.2.fastq] -> a.sam] completed Completed Task = map_dna_sequence Job = [b.sam -> b.bam] # unnecessary: already up to date Job = [c.sam -> c.bam] # unnecessary: already up to date Job = [a.sam -> a.bam] completed Completed Task = compress_sam_file Job = [b.bam -> b.statistics, use_linear_model] # unnecessary: already up to date Job = [c.bam -> c.statistics, use_linear_model] # unnecessary: already up to date Job = [a.bam -> a.statistics, use_linear_model] completed Completed Task = summarise_bam_file .. index:: pair: output_from; referring to functions before they are defined pair: output_from; defining tasks out of order .. _new_manual.output_from: *************************************** Defining pipeline tasks out of order *************************************** The examples so far assumes that all your pipelined tasks are defined in order. (``first_task`` before ``second_task``). This is usually the most sensible way to arrange your code. If you wish to refer to tasks which are not yet defined, you can do so by quoting the function name as a string and wrapping it with the :ref:`indicator class ` :ref:`output_from(...) ` so that *Ruffus* knowns this is a :term:`task` name, not a file name .. code-block:: python :emphasize-lines: 5 #--------------------------------------------------------------- # # second task # # task name string wrapped in output_from(...) @transform(output_from("first_task"), suffix(".output.1"), ".output2") def second_task(input_files, output_file): with open(output_file, "w"): pass #--------------------------------------------------------------- # # first task # @transform(first_task_params, suffix(".start"), [".output.1", ".output.extra.1"], "some_extra.string.for_example", 14) def first_task(input_files, output_file_pair, extra_parameter_str, extra_parameter_num): for output_file in output_file_pair: with open(output_file, "w"): pass #--------------------------------------------------------------- # # Run # pipeline_run([second_task]) You can also refer to tasks (functions) in other modules, in which case the full qualified name must be used: :: @transform(output_from("other_module.first_task"), suffix(".output.1"), ".output2") def second_task(input_files, output_file): pass .. index:: pair: @transform; multiple dependencies .. _new_manual.transform.multiple_dependencies: *************************************** Multiple dependencies *************************************** Each task can depend on more than one antecedent simply by chaining to a list in :ref:`@transform ` .. code-block:: python :emphasize-lines: 2 # # third_task depends on both first_task() and second_task() # @transform([first_task, second_task], suffix(".output.1"), ".output2") def third_task(input_files, output_file): with open(output_file, "w"): pass ``third_task()`` depends on and follows both ``first_task()`` and ``second_task()``. However, these latter two tasks are independent of each other and can and will run in parallel. This can be clearly shown for our example if we added a little randomness to the run time of each job: .. code-block:: python time.sleep(random.random()) The execution of ``first_task()`` and ``second_task()`` jobs will be interleaved and they finish in no particular order: .. code-block:: pycon >>> pipeline_run([third_task], multiprocess = 6) Job = [[job3.a.start, job3.b.start] -> [job3.a.output.1, job3.a.output.extra.1], some_extra.string.for_example, 14] completed Job = [[job6.a.start, job6.b.start] -> [job6.a.output.1, job6.a.output.extra.1], some_extra.string.for_example, 14] completed Job = [[job1.a.start, job1.b.start] -> [job1.a.output.1, job1.a.output.extra.1], some_extra.string.for_example, 14] completed Job = [[job4.a.start, job4.b.start] -> [job4.a.output.1, job4.a.output.extra.1], some_extra.string.for_example, 14] completed Job = [[job5.a.start, job5.b.start] -> [job5.a.output.1, job5.a.output.extra.1], some_extra.string.for_example, 14] completed Completed Task = second_task Job = [[job2.a.start, job2.b.start] -> [job2.a.output.1, job2.a.output.extra.1], some_extra.string.for_example, 14] completed .. note:: See the :ref:`example code ` .. index:: pair: @follow; imposing order with .. _new_manual.follows: *************************************** :ref:`@follows ` *************************************** If there is some extrinsic reason one non-dependent task has to precede the other, then this can be specified explicitly using :ref:`@follows `: .. code-block:: python :emphasize-lines: 2 # # @follows specifies a preceding task # @follows("first_task") @transform(second_task_params, suffix(".start"), [".output.1", ".output.extra.1"], "some_extra.string.for_example", 14) def second_task(input_files, output_file_pair, extra_parameter_str, extra_parameter_num): :ref:`@follows ` specifies either a preceding task (e.g. ``first_task``), or if it has not yet been defined, the name (as a string) of a task function (e.g. ``"first_task"``). With the addition of :ref:`@follows `, all the jobs of ``second_task()`` start *after* those from ``first_task()`` have finished: .. code-block:: pycon >>> pipeline_run([third_task], multiprocess = 6) Job = [[job2.a.start, job2.b.start] -> [job2.a.output.1, job2.a.output.extra.1], some_extra.string.for_example, 14] completed Job = [[job3.a.start, job3.b.start] -> [job3.a.output.1, job3.a.output.extra.1], some_extra.string.for_example, 14] completed Job = [[job1.a.start, job1.b.start] -> [job1.a.output.1, job1.a.output.extra.1], some_extra.string.for_example, 14] completed Completed Task = first_task Job = [[job4.a.start, job4.b.start] -> [job4.a.output.1, job4.a.output.extra.1], some_extra.string.for_example, 14] completed Job = [[job6.a.start, job6.b.start] -> [job6.a.output.1, job6.a.output.extra.1], some_extra.string.for_example, 14] completed Job = [[job5.a.start, job5.b.start] -> [job5.a.output.1, job5.a.output.extra.1], some_extra.string.for_example, 14] completed Completed Task = second_task .. index:: single: @follows; mkdir (Manual) single: mkdir; @follows (Manual) .. _new_manual.follows.mkdir: ************************************************************************************************************************************************************ Making directories automatically with :ref:`@follows ` and :ref:`mkdir ` ************************************************************************************************************************************************************ :ref:`@follows ` is also useful for making sure one or more destination directories exist before a task is run. *Ruffus* provides special syntax to support this, using the special :ref:`mkdir ` indicator class. For example: .. code-block:: python :emphasize-lines: 2 # # @follows specifies both a preceding task and a directory name # @follows("first_task", mkdir("output/results/here")) @transform(second_task_params, suffix(".start"), [".output.1", ".output.extra.1"], "some_extra.string.for_example", 14) def second_task(input_files, output_file_pair, extra_parameter_str, extra_parameter_num): Before ``second_task()`` is run, the ``output/results/here`` directory will be created if necessary. .. index:: pair: inputs parameters; globs pair: globs in input parameters; Tutorial .. _new_manual.globs_as_input: ****************************************************************************** Globs in the **Input** parameter ****************************************************************************** * As a syntactic convenience, *Ruffus* also allows you to specify a |glob|_ pattern (e.g. ``*.txt``) in the **Input** parameter. * |glob|_ patterns will be automatically specify all matching file names as the **Input**. * Any strings within **Input** which contain the letters: ``*?[]`` will be treated as a |glob|_ pattern. The first function in our initial *Ruffus* pipeline example could have been written as: .. code-block:: python :emphasize-lines: 4 # # STAGE 1 fasta->sam # @transform("*.fasta", # Input = glob suffix(".fasta"), # suffix = .fasta ".sam") # Output suffix = .sam def map_dna_sequence(input_file, output_file): "" .. index:: pair: Mixing tasks, globs and file names; Tutorial ****************************************************************************** Mixing Tasks and Globs in the **Input** parameter ****************************************************************************** |glob|_ patterns, references to tasks and file names strings can be mixed freely in (nested) python lists and tuples in the **Input** parameter. For example, a task function can chain to the **Output** from multiple upstream tasks: .. code-block:: python @transform([task1, task2, # Input = multiple tasks "aa*.fasta", + all files matching glob "zz.fasta"] + file name suffix(".fasta"), # suffix = .fasta ".sam") # Output suffix = .sam def map_dna_sequence(input_file, output_file): "" In all cases, *Ruffus* tries to do the right thing, and to make the simple or obvious case require the simplest, least onerous syntax. If sometimes *Ruffus* does not behave the way you expect, please write to the authors: it may be a bug! :ref:`new_manual.pipeline_printout` and :ref:`new_manual.cmdline` will show you how to to make sure that your intentions are reflected in *Ruffus* code.