checkpointed 0.3.0

No project description provided

Checkpointed - A library for management of nonlinear processing pipelines.

---

Introduction

The Checkpointed library provides a small interface for chaining pipelines of different operations. The main functionality of the library is the ability to maintain checkpoints for every performed operation.

---

Examples

from checkpointed import Pipeline, NoopStep

pipeline = Pipeline('example-pipeline')

# A regular input 
input1 = pipeline.add_source(
    NoopStep, name='input-1'
)

# An input whose result is also stored as an output in the file 
# 'input-2-output.txt'
input2 = pipeline.add_source(
    NoopStep, name='input-2', is_sink=True, filename='input-2-output.txt'
)

# A processing step which depends on input1 
node3 = pipeline.add_step(NoopStep, name='node-3')
pipeline.connect(input1, node3)

# A processing step which depends on both input2 and node3
node4 = pipeline.add_step(NoopStep, name='node-4')
pipeline.connect(input2, node4)
pipeline.connect(node3, node4)

# Output step which depends on node4
out5 = pipeline.add_sink(NoopStep, name='out-5', filename='out-5-output.txt')
pipeline.connect(node4, out5)

# Build the execution plan
plan = pipeline.build()

# Configuration information for all the steps.
# Empty because NoopStep does not require configuration.
config = {
    input1: {},
    input2: {},
    node3: {},
    node4: {},
    out5: {}
}

# (optional) setup a logger 
import logging

logger = logging.getLogger('example-pipeline')
logger.addHandler(logging.StreamHandler())

# Run the pipeline
plan.execute(
    # Configuration used to instantiate the steps in the pipeline.
    # Must have one entry for every step.
    config_by_step=config,

    # Store `input-2-output.txt' and `out-5-output.txt' 
    # in directory ./out/example-pipeline
    output_directory='out',

    # Store checkpoints for every pipeline step 
    # in directory ./checkpoints/example-pipeline
    checkpoint_directory='checkpoints',

    # Pass in a logger object in order to enable logging.
    logger=logger
)

With some actual arguments applied:

from checkpointed import Pipeline, NoopStep

pipeline = Pipeline('example-pipeline')

node = pipeline.add_source(NoopStep, is_sink=True, filename='out.txt')

plan = pipeline.build()

import logging

logger = logging.getLogger('example-pipeline')
logger.addHandler(logging.StreamHandler())

plan.execute(
    config_by_step={
        node: {'echo-io': True}
    },
    output_directory='out',
    checkpoint_directory='checkpoints',
    logger=logger
)

You will get an error on an invalid argument:

from checkpointed import Pipeline, NoopStep

pipeline = Pipeline('example-pipeline')

node = pipeline.add_source(NoopStep, is_sink=True, filename='out.txt')

plan = pipeline.build()

import logging

logger = logging.getLogger('example-pipeline')
logger.addHandler(logging.StreamHandler())

plan.execute(
    config_by_step={
        node: {'echo-execute': True, 'echo-io': False}
    },
    output_directory='out',
    checkpoint_directory='checkpoints',
    logger=logger
)