robdmc / consecution Goto Github PK

Hi Rob,

I am highly impressed by this library, and have intentions of creating a framework based on it.
I see your "suggestions" for other libraries where you show examples of data science oriented uses (such including pandas and dask), majorly concentrating on passing a dataframe etc. However, in consecution you have far more flexibility with any kind of approach.

The "niche" I referred to, is about being able to pass ANYTHING into the pipeline, such as passing an object instance, thus being able to add to it (configs, results, other instances, etc.). This allows building of tools with re-usable steps, and having various chains joined together to form your workflows, and building libraries of them.

I would love to see this library cherish and staying alive.

Kind Regards,
Denis.

Trying to install the library on python 3.8 and a error occurs. Checking the setup.py it seems that the latest python version that is supported is 3.5

At the moment you need to hard code the node names from the pipeline in functions used for routing. For example (from the documentation):

def route_func(item):
    if item % 2 == 0:
        return 'even_node'
    else:
        return 'odd_node'

I created a helper function to get around this issue, but feel there must be a better way...

def route_decision(decision_function, yes_node, no_node, **kwargs):
    router_function = lambda item: decision_function(item, yes_node.name, no_node.name, **kwargs)
    return [yes_node, no_node, router_function]

This enables the function above to be re-written like:

def route_func(item, yes_node_name, no_node_name):
    if item % 2 == 0:
        return yes_node_name
    else:
        return no_node_name

It would be ideal if the framework didn't require any hard coding of the node names.

For example, input is a list of 1000 elements. Flow is as Input -> Node1 -> Node2...
I hope after Node1 process 100 elements, Node2 can start to process 100 elements. And at the same time, Node1 process next 100 element.
Many thanks if some suggestions given.

The last successful build was #737 on 10/23/18. Now, every build fails during the flake8 command:

The command "flake8 ." failed and exited with 1 during .

Taking a look at the output of the pip install command, we can see different versions for a few of the dependencies:

Success (10/23/18):
Successfully installed chardet-3.0.4 consecution coverage-4.5.1 coveralls-1.5.1 docopt-0.6.2 flake8-3.5.0 graphviz-0.10idna-2.7 mccabe-0.6.1pycodestyle-2.3.1 pyflakes-1.6.0 requests-2.20.0 urllib3-1.24

Fail (10/24/18):
Successfully installed chardet-3.0.4 consecution coverage-4.5.1 coveralls-1.5.1 docopt-0.6.2 flake8-3.6.0 graphviz-0.10.1idna-2.7 mccabe-0.6.1pycodestyle-2.4.0 pyflakes-2.0.0 requests-2.20.0 urllib3-1.24

The Travis CI docs suggest locking in the dependencies to the last working version, so maybe the solution would be to explicitly set the versions for the dependencies in setup.py here.

Hi @robdmc ,

Is there a known strategy for collecting node's start and end times ?
I tried the approach below but the measured time includes the subsequent nodes processing time.

class BaseNode(Node):
    """Base Node for centralized instrumentation"""
    __metaclass__ = ABCMeta

    def __init__(self, name, **kwargs):
        super(BaseNode, self ).__init__(name, **kwargs)

    def process(self, item):
        try:
            self.clock.start(item.id)
            self.do_process(item)
        finally:
            self.clock.stop(item.id)

    @abstractmethod
    def do_process(self, item):
        pass

Ideas ? Suggestions ?

Hey there, thanks for putting this library together. I know it is no longer maintained, but I thought the approach was neat and decided to start building off of it. It turned into Glide. The syntax is similar, but quite a few new features and builtin nodes were added. Check out the docs for more info, and let me know if you give it a spin. Thanks.

Can it run pipelines and nodes on different machines?

If yes, where is the data stored? some remotely accessed database? or filesystem?