在Python中处理简单的工作流程

我正在研究一个需要数据集并在其上运行一些算法的代码。在Python中处理简单的工作流程

用户上传的数据集，然后选择哪些算法将在此数据集上运行，并创建一个像这样的工作流程：

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

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

我正在用Python写这篇文章。你能提出一些处理这个工作流程的策略吗？

来源

2010-01-24 Stephen T.

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

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这就像一个化妆工具，但进步的支持和漂亮的流程图）

来源

2010-01-24 11:41:21 Tobu

非常优雅。 – 2010-01-24 11:43:34

对不起，作为新手，但我不能让代码工作。我会从algo_by_name函数返回算法名称吗？ – 2010-01-24 16:39:48

algo_by_name（aname）需要是一个函数，所以你可以传递（p，data）给它。我写了一个例子。 – Tobu 2010-01-24 17:15:55

您想要做到这一点的方式听起来对我来说很重要，或者您需要发布更多关于您正在尝试完成的信息。

和建议：我会把工作流结构与元组的列表，而不是一本字典

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

来源

2010-01-24 11:31:04 fabrizioM

如果每个algorithm作品上dataset每个元素，map()将是一个优雅的选择：

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

eg为平方根只有奇数，使用，

>>> 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]

来源

2010-01-24 11:31:13

可能比得分最高的答案更简洁优雅，但我倾向于更倾向于这种可读性和其他用户的进一步使用。 – Morlock 2010-01-24 19:40:54

定义一个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，算法）（* PARAMS）
这将呼吁myDataset与功能名称由“算法”指定，参数列表包含在“params”中。

来源

2010-01-24 11:37:23 Sapph

这里我如何做到这一点（所有代码未经测试）：

步骤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

也许你想指定一些其他的方式的步骤。例如，一个文本文件或一个字符串（听起来可能是基于Web的？）。然后，您需要在算法上进行注册。最简单的方法是创建一个名为“registry.py”的模块，如下所示：

algorithms = {}

很简单，是吗？你会注册一个新的算法，像这样：

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

您还会需要创建从一个文本文件或东西规格链的方法。我会留给你的。 ;）

（我可能会使用Zope组件体系结构并制作算法组件，并将它们注册到组件注册表中，但这一切都严格来说是过度的）。

来源

2010-01-24 12:19:14

在Python中处理简单的工作流程

回答

相关问题