Procesamiento de un flujo de trabajo simple en Python

Question

Estoy trabajando en un código que toma un conjunto de datos y ejecuta algunos algoritmos en él.

usuario carga un conjunto de datos, y luego selecciona qué algoritmos se ejecutan en esta base de datos y crea un flujo de trabajo de esta manera:

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

Lo que significa que me quedo con workflow[0] como mi conjunto de datos, y se ejecutará algorithm1 en eso. Luego, tomaré sus resultados y ejecutaré algorithm2 en este resultado como mi nuevo conjunto de datos. Y tomaré los nuevos resultados y ejecutaré algorithm3 en él. Sigue así hasta el último elemento y no hay límite de longitud para este flujo de trabajo.

Escribo esto en Python. ¿Puede sugerir algunas estrategias sobre el procesamiento de este flujo de trabajo?

Answer 1

Quiere ejecutar una canalización en algún conjunto de datos. Eso suena como una operación de reducción (doblar en algunos idiomas).No hay necesidad de nada complicado:

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

Esto supone que el flujo de trabajo se parece a (texto orientado para que pueda cargar con YAML/JSON):

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

Y que sus algoritmos parezca:

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

algo_by_name toma un nombre y le da una función de algo; por ejemplo:

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

(antigua edición:.. si quieres un marco para la escritura de las tuberías, se puede usar Ruffus Es como una herramienta de marca, pero con el apoyo progreso y diagramas de flujo bonitos)

Answer 2

Me parece bien la forma en que quiere hacerlo, o necesita publicar más información acerca de lo que está tratando de lograr.

y asesoramiento: Yo pondría la estructura de flujo de trabajo en una lista de tuplas en lugar de un diccionario

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

Answer 3

Si cada algorithm trabajos sobre cada elemento de dataset, map() serían una opción elegante:

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

por ejemplo de las raíces cuadradas de sólo números impares, uso,

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

definir una clase Dataset que rastrea los datos ... ... para su conjunto. Definir métodos en esta clase. Algo como esto:

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 

Ahora, itere sobre su dictado de "flujo de trabajo". Para la primera entrada, simplemente instancia tu clase Dataset.

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

de cada entrada sucesiva ...

• extraer la clave/valor de alguna manera (me recomiendan cambiar la estructura de datos de flujo de trabajo, si es posible, dict es un inconveniente aquí)
• Analizar el parámetro cadena a una tupla de argumentos (este paso depende de usted).

• Suponiendo que ahora tiene la cadena algorithm y la tupla params para la iteración actual ...

  GetAttr (MyDataSet, algoritmo) (* params)

• Esto llamará a la función de myDataset con el nombre especificado por "algoritmo" con la lista de argumentos contenida en "params".

Answer 5

Aquí es cómo lo haría (todo el código no probado):

Paso 1: Necesita crear los algoritmos. El conjunto de datos podría tener este aspecto:

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

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

Tenga en cuenta que usted hace un iterador fuera de él, por lo que iterar sobre ella un elemento a la vez. Hay una razón para eso, se verá más adelante:

Otro algoritmo podría tener este aspecto:

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 

Paso entonces necesitaría 2

Su usuario para hacer una cadena de esto de alguna manera. Ahora bien, si él tenía acceso a Python directamente, puede simplemente hacer esto:

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

y luego obtener los resultados por:

for x in multiplier: 
    print x 

y sería pedir al multiplicador para una pieza de datos a la vez , y el multiplicador a su vez como el conjunto de datos. Si tiene una cadena, esto significa que se maneja una pieza de datos a la vez. Esto significa que puede manejar grandes cantidades de datos sin usar mucha memoria.

Paso 3

Probablemente desea especificar los pasos de alguna otra manera. Por ejemplo, un archivo de texto o una cadena (parece que esto puede estar basado en la web?). Entonces necesitas un registro sobre los algoritmos. La manera más fácil es simplemente crear un módulo llamado "registry.py" como este:

algorithms = {} 

Fácil, ¿eh? Se podría registrar un nuevo algoritmo de este modo:

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

También había necesidad de un método que crea la cadena de especificaciones en un archivo de texto o algo así. Te dejaré eso a ti. ;)

(Probablemente usaría la arquitectura Zope Component y haría los componentes de los algoritmos y los registraría en el registro de componentes. Pero eso es, en sentido estricto, excesivo).