Обработка простого рабочего процесса в Python

Question

Я работаю над кодом, который принимает набор данных и запускает на нем несколько алгоритмов.

Пользователь загружает набор данных, а затем выбирает, какие алгоритмы будут запускаться в этом наборе данных, и создает рабочий процесс, подобный этому:

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

Что означает, что я возьму workflow[0] в качестве своего набора данных, и я запусту algorithm1 на нем. Затем я возьму его результаты и буду использовать algorithm2 для этих результатов в качестве моего нового набора данных. И я возьму новые результаты и запусту algorithm3. Так продолжается до последнего элемента, и для этого рабочего процесса нет ограничения по длине.

Я пишу это на Python. Можете ли вы предложить некоторые стратегии обработки этого рабочего процесса?

Answer 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 . Это как инструмент make, но с поддержкой прогресса и симпатичными блок-схемами.)

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

Answer 3

Мне кажется, что то, как вы хотите это сделать, или вам нужно опубликовать больше информации о том, чего вы пытаетесь достичь.

И совет: я бы поместил структуру рабочего процесса в список с кортежами, а не в словарь

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

Answer 4

Вот как я бы это сделал (весь код не проверен):

Шаг 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, создавал компоненты алгоритмов и регистрировал их в реестре компонентов. Но это, строго говоря, перебор).

Answer 5

Определите класс 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 с именем, указанным «алгоритмом» со списком аргументов, содержащимся в «параметрах».