Skip to main content

A lightweight framework for parallel data processing using directed acyclic graphs

Project description

Pypeworks

Pypeworks is an open-source framework for implementing parallelized dataflows in Python.

Install

Pypeworks is available through the PyPI repository and can be installed using pip:

pip install pypeworks

Quick start

Pypeworks' central concept is that of the Pipework. A pipework may be defined as a directed acylic graph, wherein each Node serves as a processing unit, taking in data, transforming it, and forwarding it to the next node, until the exit node is reached.

Pypeworks offers two ways to set-up a Pipework. A pipework may be instantiated and constructed on the fly:

from pypeworks import (
    Pipework, 
    Node,
    Connection
)

pipework = (

   Pipework(

      min = Node(
         lambda xs: ("min", min(xs))
      ),

      mean = Node(
         lambda xs: ("mean", sum(xs) / len(xs))
      ),

      max = Node(
         lambda xs: ("max", max(xs))
      ),

      connections = [
         Connection( "enter" , "min"  ),
         Connection( "enter" , "mean" ),
         Connection( "enter" , "max"  ),
         Connection( "min"   , "exit" ),
         Connection( "mean"  , "exit" ),
         Connection( "max"   , "exit" )
      ]

   )

)

print(
   dict(
      pipework([1, 2, 3, 4, 5, 6, 7, 8])
   )
) # {'max': 8, 'mean': 4.5, 'min': 1}

Alternatively, a pipework may be implemented as a templatable, reusable class:

from pypeworks import (
   Pipework
)

class Pipework(Pipework):

   @Pipework.connect(input = "enter")
   @Pipework.connect(output = "exit")
   def min(self, xs : list[int]):
      return ("min", min(xs))

   @Pipework.connect(input = "enter")
   @Pipework.connect(output = "exit")
   def mean(self, xs : list[int]):
      return ("mean", sum(xs) / len(xs))

   @Pipework.connect(input = "enter")
   @Pipework.connect(output = "exit")
   def max(self, xs : list[int]):
      return ("max", max(xs))

print(
   dict(
      Pipework()([1, 2, 3, 4, 5, 6, 7, 8])
   )
) # {'max': 8, 'mean': 4.5, 'min': 1}

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

pypeworks-0.3.0.tar.gz (39.7 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

pypeworks-0.3.0-py3-none-any.whl (46.3 kB view details)

Uploaded Python 3

File details

Details for the file pypeworks-0.3.0.tar.gz.

File metadata

  • Download URL: pypeworks-0.3.0.tar.gz
  • Upload date:
  • Size: 39.7 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.1.0 CPython/3.12.0

File hashes

Hashes for pypeworks-0.3.0.tar.gz
Algorithm Hash digest
SHA256 d51cb2c2b4afef635c3fd087c25e921e78fd63a295d10fbc0c3eb746547ea5d6
MD5 d54093bba08851b6dfe2cab0e61ad299
BLAKE2b-256 49cf782435ebca2e7ac2bf46971ebcde90c9ebf56a8bc0ad29f589b57d0f4e33

See more details on using hashes here.

File details

Details for the file pypeworks-0.3.0-py3-none-any.whl.

File metadata

  • Download URL: pypeworks-0.3.0-py3-none-any.whl
  • Upload date:
  • Size: 46.3 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.1.0 CPython/3.12.0

File hashes

Hashes for pypeworks-0.3.0-py3-none-any.whl
Algorithm Hash digest
SHA256 e4cc6baa8a021fd9b5518dd0abf2fb9925e5fba2c0d7f15085d69857faa694f8
MD5 a9e76f88f42b2d91d17817433537a03f
BLAKE2b-256 f932e4a131be6051ce1e4bb4146cc917849595a745cd9ccb44f3c0df853d94b5

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page