# Chapter 12: Splitting up large tasks / files with @split

See also

- Manual Table of Contents

- @split syntax

- Example code for this chapter

## Overview

> A common requirement in computational pipelines is to split up a large task into small jobs which can be run on different processors, (or sent to a computational cluster). Very often, the number of jobs depends dynamically on the size of the task, and cannot be known beforehand.

> Ruffus uses the @split decorator to indicate that the task function will produce an indeterminate number of independent Outputs from a single Input.

## Example: Calculate variance for a large list of numbers in parallel

> Suppose we wanted to calculate the variance for 100,000 numbers, how can we parallelise the calculation so that we can get an answer as speedily as possible?

> We need to

> > - break down the problem into manageable chunks

> > - solve these in parallel, possibly on a computational cluster and then

> > - merge the partial solutions back together for a final result.

> To complicate things, we usually do not want to hard-code the number of parallel chunks beforehand. The degree of parallelism is often only apparent as we process our data.

> Ruffus was designed to solve such problems which are common, for example, in bioinformatics and genomics.

> A flowchart for our variance problem might look like this:

> (In this toy example, we create our own starting data in `create_random_numbers()`.)

## Output files for @split

> The Ruffus decorator @split is designed specifically with this run-time flexibility in mind:

>     @split(create_random_numbers, "*.chunks")
>     def split_problem (input_file_names, output_files):
>         pass

> This will split the incoming `input_file_names` into `NNN` number of outputs where `NNN` is not predetermined:

> The output (second) parameter of @split often contains a glob pattern like the `*.chunks` above.

> Only after the task function has completed, will Ruffus match the Output parameter (`*.chunks`) against the files which have been created by `split_problem()` (e.g. `1.chunks`, `2.chunks`, `3.chunks`)

## Be careful in specifying Output globs

> Note that it is your responsibility to keep the Output specification tight enough so that Ruffus does not pick up extraneous files.

> You can specify multiple glob patterns to match all the files which are the result of the splitting task function. These can even cover different directories, or groups of file names. This is a more extreme example:

> >     @split("input.file", ['a*.bits', 'b*.pieces', 'somewhere_else/c*.stuff'])
> >     def split_function (input_filename, output_files):
> >         "Code to split up 'input.file'"

## Clean up previous pipeline runs

> Problem arise when the current directory contains results of previous pipeline runs.

> - For example, if the previous analysis involved a large data set, there might be 3 chunks: `1.chunks`, `2.chunks`, `3.chunks`.

> - In the current analysis, there might be a smaller data set which divides into only 2 chunks, `1.chunks` and `2.chunks`.

> - Unfortunately, `3.chunks` from the previous run is still hanging around and will be included erroneously by the glob `*.chunks`.

> Warning

> Your first duty in @split tasks functions should be to clean up

> To help you clean up thoroughly, Ruffus initialises the output parameter to all files which match specification.

> The first order of business is thus invariably to cleanup ( delete with `os.unlink`) all files in Output.

>     #---------------------------------------------------------------
>     #
>     #   split initial file
>     #
>     @split(create_random_numbers, "*.chunks")
>     def split_problem (input_file_names, output_files):
>         """
>             splits random numbers file into xxx files of chunk_size each
>         """
>         #
>         #   clean up any files from previous runs
>         #
>         #for ff in glob.glob("*.chunks"):
>         for ff in input_file_names:
>             os.unlink(ff)

> (The first time you run the example code, `*.chunks` will initialise `output_files` to an empty list. )

## 1 to many

> @split is a one to many operator because its outputs are a list of independent items.

> If @split generates 5 files, then this will lead to 5 jobs downstream.

> This means we can just connect our old friend @transform to our pipeline and the results of @split will be analysed in parallel. This code should look familiar:

> >     #---------------------------------------------------------------
> >     #
> >     #   Calculate sum and sum of squares for each chunk file
> >     #
> >     @transform(split_problem, suffix(".chunks"), ".sums")
> >     def sum_of_squares (input_file_name, output_file_name):
> >         pass

> Which results in output like this:

> >     >>> pipeline_run()
> >         Job  = [[random_numbers.list] -> *.chunks] completed
> >     Completed Task = split_problem
> >         Job  = [1.chunks -> 1.sums] completed
> >         Job  = [10.chunks -> 10.sums] completed
> >         Job  = [2.chunks -> 2.sums] completed
> >         Job  = [3.chunks -> 3.sums] completed
> >         Job  = [4.chunks -> 4.sums] completed
> >         Job  = [5.chunks -> 5.sums] completed
> >         Job  = [6.chunks -> 6.sums] completed
> >         Job  = [7.chunks -> 7.sums] completed
> >         Job  = [8.chunks -> 8.sums] completed
> >         Job  = [9.chunks -> 9.sums] completed
> >     Completed Task = sum_of_squares

> Have a look at the Example code for this chapter

## Nothing to many

> Normally we would use @originate to create files from scratch, for example at the beginning of the pipeline.

> However, sometimes, it is not possible to determine ahead of time how many files you will be creating from scratch. @split can also be useful even in such cases:

> >     from random import randint
> >     from ruffus import *
> >     import os

> >     # Create between 2 and 5 files
> >     @split(None, "*.start")
> >     def create_initial_files(no_input_file, output_files):
> >         # cleanup first
> >         for oo in output_files:
> >             os.unlink(oo)
> >         # make new files
> >         for ii in range(randint(2,5)):
> >             open("%d.start" % ii, "w")

> >     @transform(create_initial_files, suffix(".start"), ".processed")
> >     def process_files(input_file, output_file):
> >         open(output_file, "w")

> >     pipeline_run()

> > Giving:

> >     >>> pipeline_run()
> >         Job  = [None -> *.start] completed
> >     Completed Task = create_initial_files
> >         Job  = [0.start -> 0.processed] completed
> >         Job  = [1.start -> 1.processed] completed
> >     Completed Task = process_files