Skip to content

Pipeline development framework, easy to experiment and compare different pipelines, quick to deploy to workflow orchestration tools

License

Notifications You must be signed in to change notification settings

trducng/theflow

Repository files navigation

Pipeline development framework, easy to experiment and compare different pipelines, quick to deploy to workflow orchestration tools


Most of the current workflow orchestrators focus on executing the already-developed pipelines in production. This library focuses on the pipeline development process. It aims to make it easy to develop pipeline, and once the user reaches a good pipeline, it aims to make it easy to export to other production-grade workflow orchestrators. Notable features:

  • Manage pipeline experiment: store pipeline run outputs, compare pipelines & visualize.
  • Support pipeline as code: allow complex customization.
  • Support pipeline as configuration - suitable for plug-and-play when pipeline is more stable.
  • Fast pipeline execution, auto-cache & run from cache when necessary.
  • Allow version control of artifacts with git-lfs/dvc/god...
  • Export pipeline to compatible workflow orchestration tools (e.g. Argo workflow, Airflow, Kubeflow...).

Install

pip install theflow

Quick start

(A code walk-through of this session is stored in examples/10-minutes-quick-start.ipynb. You can run it with Google Colab (TODO - attach the link).)

Pipeline can be defined as code. You initialize all the ops in self.initialize and route them in self.run.

from theflow import Function

# Define some operations used inside the pipeline
# Operation 1: normal class-based Python object
class IncrementBy(Function):

  x: int

  def run(self, y):
    return self.x + y

# Operation 2: normal Python function
def decrement_by_5(x):
  return x - 5

# Declare flow
class MathFlow(Function):

  increment: Function
  decrement: Function

  def run(self, x):
    # Route the operations in the flow
    y = self.increment(x)
    y = self.decrement(y)
    y = self.increment(y)
    return y

flow = MathFlow(increment=IncrementBy(x=10), decrement=decrement_by_5)

You run the pipeline by directly calling it. The output is the same object returned by self.run.

output = flow(x=5)
print(f"{output=}, {type(output)=}")      # output=5, type(output)=int

You can investigate pipeline's last run through the last_run property.

flow.last_run.id()                        # id of the last run
flow.last_run.logs()                      # list all information of each step
# [TODO] flow.last_run.visualize(path="vis.png")   # export the graph in `vis.png` file

Future features

  • Arguments management
  • Cache
    • cache by runs, organized by root task, allow reproducible
    • specify the files
    • the keys are like lru_cache, takes in the original input key, specify the cache, but the cache should be file-backed, for run-after-run execution.
    • cli command to manipulate cache
  • Compare pipeline in a result folder
  • Dynamically create reproducible config
  • Support pipeline branching and merging
  • Support single process or multi-processing pipeline running
  • Can synchronize changes in the workflow, allowing logs from different run to be compatible with each other
  • Compare different runs
    • Same cache directory
    • Compare evaluation result based on kwargs
  • CLI List runs
  • CLI Delete unnecessary runs
  • Add coverage, pre-commit, CI...

License

MIT License.

About

Pipeline development framework, easy to experiment and compare different pipelines, quick to deploy to workflow orchestration tools

Resources

License

Code of conduct

Security policy

Stars

Watchers

Forks

Packages

No packages published

Languages