diff --git a/pipeline.py b/pipeline.py index c5a16d8..df92427 100644 --- a/pipeline.py +++ b/pipeline.py @@ -2,6 +2,7 @@ import json from transport import factory import numpy as np +import time import os from multiprocessing import Process import pandas as pd @@ -76,7 +77,12 @@ class Components : columns = args['columns'] df = np.array_split(df[columns].values,PART_SIZE) qwriter = factory.instance(type='queue.QueueWriter',args={'queue':'aou.io'}) - part_index = 0 + part_index = 0 + # + # let's start n processes to listen & train this mother ... + # + #-- hopefully they learn as daemons + for _df in df: # _args['logs'] = os.sep.join([log_folder,str(part_index)]) @@ -206,6 +212,7 @@ class Components : if partition : info ['partition'] = int(partition) logger.write({"module":"generate","action":"write","info":info} ) + @staticmethod def callback(channel,method,header,stream): if stream.decode('utf8') in ['QUIT','EXIT','END'] : @@ -306,6 +313,7 @@ if __name__ == '__main__' : while len(jobs) > 0 : jobs = [job for job in jobs if job.is_alive()] + time.sleep(2) # pointer(qhandler)