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

See also

- Manual Table of Contents

- @jobs_limit syntax

- pipeline_run() syntax

- drmaa_wrapper.run_job() syntax

- Back to Chapter 16: :ref:`@subdivide tasks to run efficiently and regroup with @collate

## @subdivide and regroup with @collate example

>     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)

>     #
>     #   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)

>     #pipeline_printout(sys.stdout, verbose = 3)

>     pipeline_run(verbose = 1)

> Results in

>     >>> 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
>             Analysing /home/lg/temp/b.2.fragment -> /home/lg/temp/b.2.analysed
>         Job  = [b.2.fragment -> b.2.analysed] completed
>             Analysing /home/lg/temp/b.3.fragment -> /home/lg/temp/b.3.analysed
>         Job  = [b.3.fragment -> b.3.analysed] completed
>             Analysing /home/lg/temp/c.1.fragment -> /home/lg/temp/c.1.analysed
>         Job  = [c.1.fragment -> c.1.analysed] completed
>             Analysing /home/lg/temp/c.2.fragment -> /home/lg/temp/c.2.analysed
>         Job  = [c.2.fragment -> c.2.analysed] completed
>             Analysing /home/lg/temp/c.3.fragment -> /home/lg/temp/c.3.analysed
>         Job  = [c.3.fragment -> c.3.analysed] completed
>     Completed Task = analyse_fragments

>             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