true - Python usa el bucle while para mantener actualizados los scripts de trabajo y multiprocesar las tareas en cola

while en python 3

Question

Estoy tratando de escribir una secuencia de comandos de Python que analiza una carpeta y recopilar secuencias de comandos SQL actualizadas, y luego extraer datos automáticamente para la secuencia de comandos SQL. En el código, un bucle while escanea un nuevo archivo SQL y lo envía a la función de extracción de datos. Tengo problemas para entender cómo hacer una cola dinámica con el bucle while, pero también tengo multiproceso para ejecutar las tareas en la cola.

El siguiente código tiene el problema de que la iteración del bucle while funcionará en un trabajo largo antes de pasar a la siguiente iteración y recopilará otros trabajos para llenar el procesador vacío.

Actualizar:

1. Gracias a @pbacterio por detectar el error, y ahora el mensaje de error desaparece. Después de cambiar el código, el código de Python puede tomar todos los scripts de trabajo durante una iteración y distribuir los scripts a cuatro procesadores. Sin embargo, se quedará bloqueado por un trabajo largo para pasar a la siguiente iteración, escaneando y enviando los scripts de trabajo recién agregados. ¿Alguna idea de cómo reconstruir el código?

2. Finalmente encontré la solución, vea la respuesta a continuación. Resultó que lo que buscaba es

   the_queue = Queue ()
   the_pool = Pool (4, worker_main, (the_queue,))

3. Para aquellos que se topan con la idea similar, a continuación se muestra la arquitectura completa de este script de automatización que convierte una unidad compartida en un ''servidor para la extracción de SQL'' o cualquier otro ''servidor'' en la cola de trabajos.

   a. El script de python auto_data_pull.py como se muestra en la respuesta. Necesitas agregar tu propia función de trabajo.

   segundo. Un ''script por lotes'' con lo siguiente:

   iniciar C: / Anaconda2 / python.exe C: / Users / bin / auto_data_pull.py

   do. Agregue una tarea activada por la computadora de inicio, ejecute el ''script por lotes'' Eso es todo. Funciona.

Código Python:

from glob import glob
import os, time
import sys
import CSV
import re
import subprocess
import pandas as PD
import pypyodbc
from multiprocessing import Process, Queue, current_process, freeze_support

#
# Function run by worker processes
#

def worker(input, output):
    for func, args in iter(input.get, ''STOP''):
        result = compute(func, args)
        output.put(result)

#
# Function used to compute result
#

def compute(func, args):
    result = func(args)
    return ''%s says that %s%s = %s'' % /
        (current_process().name, func.__name__, args, result)


def query_sql(sql_file): #test func
    #jsl file processing and SQL querying, data table will be saved to csv.
    fo_name = os.path.splitext(sql_file)[0] + ''.csv''
    fo = open(fo_name, ''w'')
    print sql_file
    fo.write("sql_file {0} is done/n".format(sql_file))
    return "Query is done for /n".format(sql_file)


def check_files(path):
    """
    arguments -- root path to monitor
    returns   -- dictionary of {file: timestamp, ...}
    """
    sql_query_dirs = glob(path + "/*/IDABox/")

    files_dict = {}
    for sql_query_dir in sql_query_dirs:
        for root, dirs, filenames in os.walk(sql_query_dir):
            [files_dict.update({(root + filename): os.path.getmtime(root + filename)}) for 
                     filename in filenames if filename.endswith(''.jsl'')]
    return files_dict


##### working in single thread
def single_thread():
    path = "Y:/"

    before = check_files(path)
    sql_queue  = [] 

    while True:
        time.sleep(3)
        after = check_files(path)
        added = [f for f in after if not f in before]
        deleted = [f for f in before if not f in after]
        overlapped = list(set(list(after)) & set(list(before)))
        updated = [f for f in overlapped if before[f] < after[f]]  

        before = after

        sql_queue = added + updated
        # print sql_queue
        for sql_file in sql_queue:
            try:
                query_sql(sql_file)
            except:
                pass


##### not working in queue
def multiple_thread():

    NUMBER_OF_PROCESSES = 4
    path = "Y:/"

    sql_queue  = [] 
    before = check_files(path) # get the current dictionary of sql_files
    task_queue = Queue()
    done_queue = Queue()

    while True:         #while loop to check the changes of the files
        time.sleep(5)
        after = check_files(path)
        added = [f for f in after if not f in before]
        deleted = [f for f in before if not f in after]
        overlapped = list(set(list(after)) & set(list(before)))
        updated = [f for f in overlapped if before[f] < after[f]]  

        before = after  
        sql_queue = added + updated   

        TASKS = [(query_sql, sql_file) for sql_file in sql_queue]
        # Create queues

        #submit task
        for task in TASKS:
            task_queue.put(task)

        for i in range(NUMBER_OF_PROCESSES):
                p = Process(target=worker, args=(task_queue, done_queue)).start()          
            # try:
            #     p = Process(target=worker, args=(task_queue))
            #     p.start()

            # except:
            #     pass 

        # Get and print results
        print ''Unordered results:''
        for i in range(len(TASKS)):
            print ''/t'', done_queue.get()
        # Tell child processes to stop
        for i in range(NUMBER_OF_PROCESSES):
            task_queue.put(''STOP'')        

# single_thread()
if __name__ == ''__main__'':
    # freeze_support()
    multiple_thread()

Referencia:

1. monitorea los cambios en el archivo con el script de Python: http://timgolden.me.uk/python/win32_how_do_i/watch_directory_for_changes.html
2. Multiprocesamiento:
   https://docs.python.org/2/library/multiprocessing.html

Answer 1

¿Dónde sql_file en multiple_thread() en

multiprocessing.Process(target=query_sql, args=(sql_file)).start()

No ha definido sql_file en el método y además ha usado esa variable en un bucle for. El alcance de la variable solo se limita al bucle for.

Answer 2

He resuelto esto Gracias por la respuesta inspiró el pensamiento. Ahora el script puede ejecutar un bucle while para monitorear la carpeta en busca de un nuevo script SQL actualizado / agregado, y luego distribuir la extracción de datos a múltiples hilos. La solución proviene de queue.get () y queue.put (). Asumo que el objeto de cola se encarga de la comunicación por sí mismo.

Este es el código final -

from glob import glob
import os, time
import sys
import pypyodbc
from multiprocessing import Process, Queue, Event, Pool, current_process, freeze_support

def query_sql(sql_file): #test func
    #jsl file processing and SQL querying, data table will be saved to csv.
    fo_name = os.path.splitext(sql_file)[0] + ''.csv''
    fo = open(fo_name, ''w'')
    print sql_file
    fo.write("sql_file {0} is done/n".format(sql_file))
    return "Query is done for /n".format(sql_file)


def check_files(path):
    """
    arguments -- root path to monitor
    returns   -- dictionary of {file: timestamp, ...}
    """
    sql_query_dirs = glob(path + "/*/IDABox/")

    files_dict = {}
    try:
        for sql_query_dir in sql_query_dirs:
            for root, dirs, filenames in os.walk(sql_query_dir):
                [files_dict.update({(root + filename): os.path.getmtime(root + filename)}) for 
                         filename in filenames if filename.endswith(''.jsl'')]
    except:
        pass

    return files_dict


def worker_main(queue):
    print os.getpid(),"working"
    while True:
        item = queue.get(True)
        query_sql(item)

def main():
    the_queue = Queue()
    the_pool = Pool(4, worker_main,(the_queue,))

    path = "Y:/"
    before = check_files(path) # get the current dictionary of sql_files
    while True:         #while loop to check the changes of the files
        time.sleep(5)
        sql_queue  = [] 
        after = check_files(path)
        added = [f for f in after if not f in before]
        deleted = [f for f in before if not f in after]
        overlapped = list(set(list(after)) & set(list(before)))
        updated = [f for f in overlapped if before[f] < after[f]]  

        before = after  
        sql_queue = added + updated   
        if sql_queue:
            for jsl_file in sql_queue:
                try:
                    the_queue.put(jsl_file)
                except:
                    print "{0} failed with error {1}. /n".format(jsl_file, str(sys.exc_info()[0]))
                    pass
        else:
            pass

if __name__ == "__main__":
    main()  

Answer 3

Intenta reemplazar esto:

result = func(*args)

por esto:

result = func(args)