¿Cuál es la mejor manera de dividir archivos grandes en Python para multiprocesamiento?

Question

Me encuentro con muchos "vergonzosamente paralelos" proyectos que me gustaría paralelizar con el módulo multiprocessing . Sin embargo, a menudo implican la lectura en archivos grandes (mayores de 2 gb), procesándolos línea por línea, ejecutando cálculos básicos y luego escribiendo resultados. ¿Cuál es la mejor manera de dividir un archivo y procesarlo utilizando el módulo de multiprocesamiento de Python? ¿Deben usarse Queue o JoinableQueue en multiprocessing ? ¿O el módulo Queue en sí mismo? O, ¿debería asignar el archivo iterable sobre un grupo de procesos usando multiprocessing ? He experimentado con estos enfoques, pero la sobrecarga es inmensa en la distribución de los datos línea por línea. Me decidí por un diseño liviano de filtros de tubería usando cat file | proceso1 --out-file out1 --num-procesos 2 | process2 --out-file out2 , que pasa un cierto porcentaje de la entrada del primer proceso directamente a la segunda entrada (consulte esta publicación ), pero me gustaría tener una solución contenida completamente en Python.

Sorprendentemente, la documentación de Python no sugiere una forma canónica de hacerlo (a pesar de una larga sección sobre pautas de programación en la documentación de multiprocesamiento ).

Gracias Vince

Información adicional: El tiempo de procesamiento por línea varía. Algunos problemas son rápidos y casi no están vinculados a E / S, algunos están vinculados a la CPU. Las tareas no dependientes vinculadas a la CPU obtendrán la publicación de la paralelización, de modo que incluso las formas ineficientes de asignar datos a una función de procesamiento seguirían siendo beneficiosas en términos de tiempo de reloj de pared.

Un buen ejemplo es un script que extrae campos de líneas, verifica una variedad de marcas bit a bit y escribe líneas con ciertas banderas en un archivo nuevo en un formato completamente nuevo. Esto parece un problema relacionado con E / S, pero cuando lo ejecuté con mi versión concurrente barata con tuberías, fue aproximadamente un 20% más rápido. Cuando lo ejecuto con grupo y mapa, o hago cola en multiprocessing , siempre es un 100% más lento.

Answer 1

Una de las mejores arquitecturas ya forma parte de los sistemas operativos Linux. No se requieren bibliotecas especiales.

Desea un " abanico " diseño.

1. A " principal " El programa crea una serie de subprocesos conectados por tuberías.

2. El programa principal lee el archivo, escribe líneas en las tuberías haciendo el filtrado mínimo requerido para distribuir las líneas a los subprocesos apropiados.

Cada subproceso probablemente debería ser una tubería de procesos distintos que leen y escriben desde stdin.

No necesita una estructura de datos de cola, eso es exactamente lo que es una canalización en memoria: una cola de bytes entre dos procesos concurrentes.

Answer 2

Una estrategia es asignar a cada trabajador un desplazamiento, de modo que si tiene ocho procesos de trabajador, asigne los números del 0 al 7. El número de trabajador 0 lee el primer proceso de registro, luego omite el 7 y pasa a procesar el octavo registro, etc. el trabajador número 1 lee el segundo registro, luego omite 7 y procesa el noveno registro .........

Hay una serie de ventajas para este esquema. No importa cuán grande sea el archivo, el trabajo siempre se divide de manera uniforme, los procesos en la misma máquina procesarán aproximadamente a la misma velocidad y usarán las mismas áreas de almacenamiento intermedio para que no incurra en una sobrecarga de E / S excesiva. Mientras el archivo no se haya actualizado, puede volver a ejecutar subprocesos individuales para recuperarse de los errores.

Answer 3

No mencionas cómo estás procesando las líneas; posiblemente la pieza de información más importante.

¿Cada línea es independiente? ¿El cálculo depende de una línea que viene antes de la siguiente? ¿Deben procesarse en bloques? ¿Cuánto dura el procesamiento de cada línea? ¿Hay un paso de procesamiento que debe incorporar "todos" los datos al final? ¿O se pueden tirar los resultados intermedios y mantener solo un total acumulado? ¿Se puede dividir inicialmente el archivo dividiendo el tamaño del archivo por el recuento de hilos? ¿O crece a medida que lo procesas?

Si las líneas son independientes y el archivo no crece, la única coordinación que necesita es cultivar " direcciones de inicio " y "longitudes" a cada uno de los trabajadores; pueden abrirse independientemente y buscar en el archivo y luego simplemente debe coordinar sus resultados; quizás esperando que los resultados de N vuelvan a la cola.

Si las líneas no son independientes, la respuesta dependerá en gran medida de la estructura del archivo.

Answer 4

Sé que preguntaste específicamente sobre Python, pero te animaré a mirar Hadoop ( http: // hadoop. apache.org/ ): implementa el algoritmo de Mapa y Reducir que fue diseñado específicamente para abordar este tipo de problema.

Buena suerte

Answer 5

Depende mucho del formato de su archivo.

¿Tiene sentido dividirlo en alguna parte? ¿O necesita dividirlo en una nueva línea? ¿O necesita asegurarse de dividirlo al final de una definición de objeto?

En lugar de dividir el archivo, debe usar varios lectores en el mismo archivo, usando os.lseek para saltar a la parte apropiada del archivo.

Actualización: El póster agregó que quiere dividirse en nuevas líneas. Entonces propongo lo siguiente:

Digamos que tienes 4 procesos. Entonces, la solución simple es os.lseek al 0%, 25%, 50% y 75% del archivo, y leer bytes hasta llegar a la primera línea nueva. Ese es su punto de partida para cada proceso. No necesita dividir el archivo para hacer esto, solo busque la ubicación correcta en el archivo grande en cada proceso y comience a leer desde allí.

Answer 6

Fredrik Lundh Algunas notas sobre el Benchmark de buscador amplio de Tim Bray es una lectura interesante, sobre un caso de uso muy similar, con muchos buenos consejos. Varios otros autores también implementaron lo mismo, algunos están vinculados desde el artículo, pero es posible que desee intentar buscar en Google '' python wide finder '' o algo para encontrar algo más. (también había una solución basada en el módulo multiprocesamiento , pero parece que ya no está disponible)

Answer 7

Si el tiempo de ejecución es largo, en lugar de hacer que cada proceso lea su siguiente línea a través de una Queue , haga que los procesos lean lotes de líneas. De esta forma, los gastos generales se amortizan en varias líneas (por ejemplo, miles o más).