用 Python 处理一个简单的工作流答案

【问题标题】：Processing a simple workflow in Python用 Python 处理一个简单的工作流
【发布时间】：2010-01-24 11:20:47
【问题描述】：

我正在编写一个获取数据集并在其上运行一些算法的代码。

用户上传一个数据集，然后选择将在该数据集上运行哪些算法并创建如下工作流：

workflow = 
{0: {'dataset': 'some dataset'},
 1: {'algorithm1': "parameters"},
 2: {'algorithm2': "parameters"},
 3: {'algorithm3': "parameters"}
}

这意味着我会将workflow[0] 作为我的数据集，并在其上运行algorithm1。然后，我将获取它的结果，并在这个结果上运行 algorithm2 作为我的新数据集。我将获取新结果并在其上运行algorithm3。直到最后一个项目都是这样，并且此工作流程没有长度限制。

我正在用 Python 编写这个。您能否建议一些有关处理此工作流程的策略？

【问题讨论】：

标签： python workflow

【解决方案1】：

您想在某个数据集上运行管道。这听起来像是一个减少操作（在某些语言中是折叠）。不需要任何复杂的东西：

result = reduce(lambda data, (aname, p): algo_by_name(aname)(p, data), workflow)

这假设工作流程看起来像（面向文本，因此您可以使用 YAML/JSON 加载它）：

workflow = ['data', ('algo0', {}), ('algo1', {'param': value}), … ]

你的算法看起来像：

def algo0(p, data):
    …
    return output_data.filename

algo_by_name 取一个名字并给你一个算法函数；例如：

def algo_by_name(name):
    return {'algo0': algo0, 'algo1': algo1, }[name]

（旧版本：如果你想要一个编写管道的框架，你可以使用Ruffus。它就像一个制作工具，但有进度支持和漂亮的流程图。）

【讨论】：

对不起，我是新手，但我无法使代码正常工作。我会从 algo_by_name 函数返回算法名称吗？
algo_by_name(aname) 需要是一个函数，因此您可以将 (p, data) 传递给它。我写了一个例子。
谢谢你的例子，我很感激。

【解决方案2】：

如果每个algorithm 都适用于dataset 上的每个元素，那么map() 将是一个不错的选择：

dataset=workflow[0]
for algorithm in workflow[1:]:
    dataset=map(algorithm, dataset)

例如仅对于奇数的平方根，使用，

>>> algo1=lambda x:0 if x%2==0 else x
>>> algo2=lambda x:x*x
>>> dataset=range(10)
>>> workflow=(dataset, algo1, algo2)
>>> for algo in workflow[1:]:
    dataset=map(algo, dataset)
>>> dataset
[0, 1, 0, 9, 0, 25, 0, 49, 0, 81]

【讨论】：

可能不如得分最高的答案简洁和优雅，但我更喜欢它的可读性和其他用户的进一步使用。

【解决方案3】：

我觉得你想做的方式听起来不错，或者你需要发布更多关于你想要完成的信息。

建议：我会将工作流结构放在一个包含元组而不是字典的列表中

workflow = [ ('dataset', 'some dataset'),
             ('algorithm1', "parameters"),
             ('algorithm2', "parameters"),
             ('algorithm3', "parameters")]

【讨论】：

【解决方案4】：

定义一个Dataset 类来跟踪...数据...为您的集合。在这个类中定义方法。像这样的：

class Dataset:
    # Some member fields here that define your data, and a constructor

    def algorithm1(self, param1, param2, param3):
        # Update member fields based on algorithm

    def algorithm2(self, param1, param2):
        # More updating/processing

现在，遍历您的“工作流程”字典。对于第一个条目，只需实例化您的 Dataset 类。

myDataset = Dataset() # Whatever actual construction you need to do

对于每个后续条目...

以某种方式提取键/值（如果可能，我建议更改您的工作流数据结构，dict 在这里不方便）
将参数字符串解析为参数元组（此步骤由您决定）。
假设您现在拥有当前迭代的字符串 algorithm 和元组 params...

getattr(myDataset, algorithm)(*params)
这将调用myDataset 上的函数，其名称由“算法”指定，参数列表包含在“参数”中。

【讨论】：

【解决方案5】：

我会这样做（所有代码未经测试）：

第 1 步：您需要创建算法。数据集可能如下所示：

class Dataset(object):
    def __init__(self, dataset):
        self.dataset = dataset

    def __iter__(self):
        for x in self.dataset:
            yield x

请注意，您从中创建了一个迭代器，因此您一次迭代一个项目。这是有原因的，稍后您会看到：

另一种算法可能如下所示：

class Multiplier(object):
    def __init__(self, previous, multiplier):
        self.previous = previous
        self.multiplier = multiplier
    def __iter__(self):
        for x in previous:
            yield x * self.multiplier

第 2 步

然后，您的用户将需要以某种方式创建一个链。现在，如果他可以直接访问 Python，您可以这样做：

dataset = Dataset(range(100))
multiplier = Multiplier(dataset, 5)

然后通过以下方式获取结果：

for x in multiplier:
    print x

它会一次向乘法器询问一条数据，而乘法器又会作为数据集。如果你有一个链，那么这意味着一次处理一个数据。这意味着您可以在不使用大量内存的情况下处理大量数据。

第 3 步

您可能想以其他方式指定步骤。例如文本文件或字符串（听起来可能是基于网络的？）。然后，您需要对算法进行注册表。最简单的方法是像这样创建一个名为“registry.py”的模块：

algorithms = {}

简单，嗯？您可以像这样注册一个新算法：

from registry import algorithms
algorithms['dataset'] = Dataset
algorithms['multiplier'] = Multiplier

您还需要一种从文本文件或其他文件中的规范创建链的方法。我会把它留给你。 ;)

（我可能会使用 Zope 组件架构并制作算法组件并将它们注册到组件注册表中。但严格来说，这都是矫枉过正）。

【讨论】：