Chapter 3: Python Code for More on `@transform`-ing data¶

See also

Manual Table of Contents
@transform syntax in detail
Back to Chapter 3: More on @transform-ing data and @originate

Producing several items / files per job¶

from ruffus import *

#---------------------------------------------------------------
#   Create pairs of input files
#
first_task_params = [
                     ['job1.a.start', 'job1.b.start'],
                     ['job2.a.start', 'job2.b.start'],
                     ['job3.a.start', 'job3.b.start'],
                    ]

for input_file_pairs in first_task_params:
    for input_file in input_file_pairs:
        open(input_file, "w")


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


#---------------------------------------------------------------
#
#   second task
#
@transform(first_task, suffix(".output.1"), ".output2")
def second_task(input_files, output_file):
    with open(output_file, "w"): pass


#---------------------------------------------------------------
#
#       Run
#
pipeline_run([second_task])

Resulting Output¶

>>> pipeline_run([second_task])
    Job  = [[job1.a.start, job1.b.start] -> [job1.a.output.1, job1.a.output.extra.1], some_extra.string.for_example, 14] completed
    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
Completed Task = first_task
    Job  = [[job1.a.output.1, job1.a.output.extra.1] -> job1.a.output2] completed
    Job  = [[job2.a.output.1, job2.a.output.extra.1] -> job2.a.output2] completed
    Job  = [[job3.a.output.1, job3.a.output.extra.1] -> job3.a.output2] completed
Completed Task = second_task

Defining tasks function out of order¶

 from ruffus import *

 #---------------------------------------------------------------
 #   Create pairs of input files
 #
 first_task_params = [
                      ['job1.a.start', 'job1.b.start'],
                      ['job2.a.start', 'job2.b.start'],
                      ['job3.a.start', 'job3.b.start'],
                     ]

 for input_file_pairs in first_task_params:
     for input_file in input_file_pairs:
         open(input_file, "w")



 #---------------------------------------------------------------
 #
 #   second task defined first
 #
 #   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])

Resulting Output¶

>>> pipeline_run([second_task])
    Job  = [[job1.a.start, job1.b.start] -> [job1.a.output.1, job1.a.output.extra.1], some_extra.string.for_example, 14] completed
    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
Completed Task = first_task
    Job  = [[job1.a.output.1, job1.a.output.extra.1] -> job1.a.output2] completed
    Job  = [[job2.a.output.1, job2.a.output.extra.1] -> job2.a.output2] completed
    Job  = [[job3.a.output.1, job3.a.output.extra.1] -> job3.a.output2] completed
Completed Task = second_task

Multiple dependencies¶

 from ruffus import *
 import time
 import random

 #---------------------------------------------------------------
 #   Create pairs of input files
 #
 first_task_params = [
                      ['job1.a.start', 'job1.b.start'],
                      ['job2.a.start', 'job2.b.start'],
                      ['job3.a.start', 'job3.b.start'],
                     ]
 second_task_params = [
                      ['job4.a.start', 'job4.b.start'],
                      ['job5.a.start', 'job5.b.start'],
                      ['job6.a.start', 'job6.b.start'],
                     ]

 for input_file_pairs in first_task_params + second_task_params:
     for input_file in input_file_pairs:
         open(input_file, "w")



 #---------------------------------------------------------------
 #
 #   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
     time.sleep(random.random())



 #---------------------------------------------------------------
 #
 #   second 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):
     for output_file in output_file_pair:
         with open(output_file, "w"):
             pass
     time.sleep(random.random())


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


 #---------------------------------------------------------------
 #
 #       Run
 #
 pipeline_run([third_task], multiprocess = 6)

Resulting Output¶

>>> 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
Completed Task = first_task
    Job  = [[job1.a.output.1, job1.a.output.extra.1] -> job1.a.output2] completed
    Job  = [[job2.a.output.1, job2.a.output.extra.1] -> job2.a.output2] completed
    Job  = [[job3.a.output.1, job3.a.output.extra.1] -> job3.a.output2] completed
    Job  = [[job4.a.output.1, job4.a.output.extra.1] -> job4.a.output2] completed
    Job  = [[job5.a.output.1, job5.a.output.extra.1] -> job5.a.output2] completed
    Job  = [[job6.a.output.1, job6.a.output.extra.1] -> job6.a.output2] completed
Completed Task = third_task

Multiple dependencies after @follows¶

from ruffus import *
import time
import random

#---------------------------------------------------------------
#   Create pairs of input files
#
first_task_params = [
                     ['job1.a.start', 'job1.b.start'],
                     ['job2.a.start', 'job2.b.start'],
                     ['job3.a.start', 'job3.b.start'],
                    ]
second_task_params = [
                     ['job4.a.start', 'job4.b.start'],
                     ['job5.a.start', 'job5.b.start'],
                     ['job6.a.start', 'job6.b.start'],
                    ]

for input_file_pairs in first_task_params + second_task_params:
    for input_file in input_file_pairs:
        open(input_file, "w")



#---------------------------------------------------------------
#
#   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
    time.sleep(random.random())



#---------------------------------------------------------------
#
#   second 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):
    for output_file in output_file_pair:
        with open(output_file, "w"):
            pass
    time.sleep(random.random())


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


#---------------------------------------------------------------
#
#       Run
#
pipeline_run([third_task], multiprocess = 6)

Resulting Output: `first_task` completes before `second_task`¶

>>> 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
    Job  = [[job1.a.output.1, job1.a.output.extra.1] -> job1.a.output2] completed
    Job  = [[job2.a.output.1, job2.a.output.extra.1] -> job2.a.output2] completed
    Job  = [[job3.a.output.1, job3.a.output.extra.1] -> job3.a.output2] completed
    Job  = [[job4.a.output.1, job4.a.output.extra.1] -> job4.a.output2] completed
    Job  = [[job5.a.output.1, job5.a.output.extra.1] -> job5.a.output2] completed
    Job  = [[job6.a.output.1, job6.a.output.extra.1] -> job6.a.output2] completed

Read the Docs v: latest

Versions: latest

Downloads: pdf; htmlzip; epub

On Read the Docs: Project Home; Builds

Free document hosting provided by Read the Docs.