# Chapter 16: @subdivide tasks to run efficiently and regroup with @collate

See also

- Manual Table of Contents
- @subdivide syntax
- @collate syntax

## Overview

> In Chapter 12 and Chapter 13, we saw how a large task can be @split into small jobs to be analysed efficiently in parallel. Ruffus can then @merge these back together to give a single, unified result.

> This assumes that your pipeline is processing one item at a time. Usually, however, we will have, for example, 10 large pieces of data in play, each of which has to be subdivided into smaller pieces for analysis before being put back together.

> This is the role of @subdivide and @subdivide.

> Like @split, the number of output files @subdivide produces for each Input is not predetermined.

> On the other hand, these output files should be named in such a way that they can later be grouped back together later using @subdivide.

> This will be clearer with some worked examples.

## @subdivide in parallel

> Let us start from 3 files with varying number of lines. We wish to process these two lines at a time but we do not know ahead of time how long each file is:

>     from ruffus import *
>     import os, random, sys

>     # Create files a random number of lines
>     @originate(["a.start",
>                 "b.start",
>                 "c.start"])
>     def create_test_files(output_file):
>         cnt_lines = random.randint(1,3) * 2
>         with open(output_file, "w") as oo:
>             for ii in range(cnt_lines):
>                 oo.write("data item = %d\n" % ii)
>             print "        %s has %d lines" % (output_file, cnt_lines)

>     #
>     #   subdivide the input files into NNN fragment files of 2 lines each
>     #
>     @subdivide( create_test_files,
>                 formatter(),
>                 "{path[0]}/{basename[0]}.*.fragment",
>                 "{path[0]}/{basename[0]}")
>     def subdivide_files(input_file, output_files, output_file_name_stem):
>         #
>         #   cleanup any previous results
>         #
>         for oo in output_files:
>             os.unlink(oo)
>         #
>         #   Output files contain two lines each
>         #       (new output files every even line)
>         #
>         cnt_output_files = 0
>         for ii, line in enumerate(open(input_file)):
>             if ii % 2 == 0:
>                 cnt_output_files += 1
>                 output_file_name = "%s.%d.fragment" % (output_file_name_stem, cnt_output_files)
>                 output_file = open(output_file_name, "w")
>                 print "        Subdivide %s -> %s" % (input_file, output_file_name)
>             output_file.write(line)

>     #
>     #   Analyse each fragment independently
>     #
>     @transform(subdivide_files, suffix(".fragment"), ".analysed")
>     def analyse_fragments(input_file, output_file):
>         print "        Analysing %s -> %s" % (input_file, output_file)
>         with open(output_file, "w") as oo:
>             for line in open(input_file):
>                 oo.write("analysed " + line)

> This produces the following output:

>     >>> pipeline_run(verbose = 1)
>             a.start has 2 lines
>         Job  = [None -> a.start] completed
>             b.start has 6 lines
>         Job  = [None -> b.start] completed
>             c.start has 6 lines
>         Job  = [None -> c.start] completed
>     Completed Task = create_test_files

>             Subdivide a.start -> /home/lg/temp/a.1.fragment
>         Job  = [a.start -> a.*.fragment, a] completed

>             Subdivide b.start -> /home/lg/temp/b.1.fragment
>             Subdivide b.start -> /home/lg/temp/b.2.fragment
>             Subdivide b.start -> /home/lg/temp/b.3.fragment
>         Job  = [b.start -> b.*.fragment, b] completed

>             Subdivide c.start -> /home/lg/temp/c.1.fragment
>             Subdivide c.start -> /home/lg/temp/c.2.fragment
>             Subdivide c.start -> /home/lg/temp/c.3.fragment
>         Job  = [c.start -> c.*.fragment, c] completed

>     Completed Task = subdivide_files

>             Analysing /home/lg/temp/a.1.fragment -> /home/lg/temp/a.1.analysed
>         Job  = [a.1.fragment -> a.1.analysed] completed
>             Analysing /home/lg/temp/b.1.fragment -> /home/lg/temp/b.1.analysed
>         Job  = [b.1.fragment -> b.1.analysed] completed

>         [ ...SEE EXAMPLE CODE FOR MORE LINES ...]

>     Completed Task = analyse_fragments

> `a.start` has two lines and results in a single `.fragment` file, while there are 3 `b.*.fragment` files because it has 6 lines. Whatever their origin, all of the different fragment files are treated equally in `analyse_fragments()` and processed (in parallel) in the same way.

## Grouping using @collate

> All that is left in our example is to reassemble the analysed fragments back together into 3 sets of results corresponding to the original 3 pieces of starting data.

> This is straightforward by eye: the file names all have the same pattern: `[abc].*.analysed`:

> >     a.1.analysed    ->   a.final_result
> >     b.1.analysed    ->   b.final_result
> >     b.2.analysed    ->   ..
> >     b.3.analysed    ->   ..
> >     c.1.analysed    ->   c.final_result
> >     c.2.analysed    ->   ..

> @collate does something similar:

> > 1. Specify a string substitution e.g. `c.??.analysed -> c.final_result` and
> > 2. Ask ruffus to group together any Input (e.g. `c.1.analysed`, `c.2.analysed`) that will result in the same Output (e.g. `c.final_result`)

> >     #
> >     #   ``XXX.??.analysed  -> XXX.final_result``
> >     #   Group results using original names
> >     #
> >     @collate(   analyse_fragments,

> >                 # split file name into [abc].NUMBER.analysed
> >                 formatter("/(?P<NAME>[abc]+)\.\d+\.analysed$"),

> >                 "{path[0]}/{NAME[0]}.final_result")
> >     def recombine_analyses(input_file_names, output_file):
> >         with open(output_file, "w") as oo:
> >             for input_file in input_file_names:
> >                 print "        Recombine %s -> %s" % (input_file, output_file)
> >                 for line in open(input_file):
> >                     oo.write(line)

> > This produces the following output:

> >             Recombine /home/lg/temp/a.1.analysed -> /home/lg/temp/a.final_result
> >         Job  = [[a.1.analysed] -> a.final_result] completed
> >             Recombine /home/lg/temp/b.1.analysed -> /home/lg/temp/b.final_result
> >             Recombine /home/lg/temp/b.2.analysed -> /home/lg/temp/b.final_result
> >             Recombine /home/lg/temp/b.3.analysed -> /home/lg/temp/b.final_result
> >         Job  = [[b.1.analysed, b.2.analysed, b.3.analysed] -> b.final_result] completed
> >             Recombine /home/lg/temp/c.1.analysed -> /home/lg/temp/c.final_result
> >             Recombine /home/lg/temp/c.2.analysed -> /home/lg/temp/c.final_result
> >             Recombine /home/lg/temp/c.3.analysed -> /home/lg/temp/c.final_result
> >         Job  = [[c.1.analysed, c.2.analysed, c.3.analysed] -> c.final_result] completed
> >     Completed Task = recombine_analyses

> Warning

> - Input file names are grouped together not in a guaranteed order.

>   > For example, the fragment files may not be sent to `recombine_analyses(input_file_names, ...)` in alphabetically or any other useful order.

>   > You may want to sort Input before concatenation.

> - All Input are grouped together if they have both the same Output and Extra parameters. If any string substitution is specified in any of the other Extra parameters to @subdivide, they must give the same answers for Input in the same group.