El módulo multiprocessing permite ejecutar trabajo en procesos separados y aprovechar varios núcleos de CPU. A diferencia de los hilos, cada proceso tiene su propio intérprete y su propia memoria, por lo que puede ejecutar código Python en paralelo para tareas intensivas como cálculos, compresión, procesamiento de imágenes o transformaciones numéricas. El beneficio no es automático: crear procesos y transferir datos también tiene un coste.
Antes de elegir esta herramienta, compara el enfoque con threading en Python, comprende el GIL, revisa asyncio para tareas de espera y mide resultados con timeit.
Cuándo usar multiprocessing
Es una buena opción cuando la tarea consume CPU y puede dividirse en partes relativamente independientes. Ejemplos: calcular hashes de muchos archivos, aplicar filtros a imágenes, analizar documentos o ejecutar simulaciones. Para descargar páginas, esperar respuestas de APIs o leer sockets, los hilos o asyncio suelen consumir menos recursos.
Crear un proceso básico
from multiprocessing import Process
import os
def trabajar(nombre):
print(f"Proceso {nombre}, PID {os.getpid()}")
if __name__ == "__main__":
proceso = Process(target=trabajar, args=("A",))
proceso.start()
proceso.join()
print("Proceso finalizado")start() inicia el proceso y join() espera a que termine. El bloque if __name__ == "__main__" es esencial, especialmente en Windows y macOS, para evitar que cada proceso vuelva a crear otros procesos de forma recursiva.
Ejecutar varios procesos
from multiprocessing import Process
def cuadrado(numero):
print(numero, numero ** 2)
if __name__ == "__main__":
procesos = []
for numero in range(4):
proceso = Process(target=cuadrado, args=(numero,))
proceso.start()
procesos.append(proceso)
for proceso in procesos:
proceso.join()El orden de salida no está garantizado. Los procesos avanzan según la planificación del sistema operativo.
Usar Pool para muchas tareas
Pool mantiene un conjunto de trabajadores y distribuye elementos:
from multiprocessing import Pool
def calcular(numero):
return numero ** 2
if __name__ == "__main__":
numeros = list(range(10))
with Pool() as pool:
resultados = pool.map(calcular, numeros)
print(resultados)map() conserva el orden de entrada. El gestor de contexto cierra y espera a los trabajadores correctamente.
Elegir el número de procesos
Puedes consultar los núcleos lógicos:
import os
print(os.cpu_count())Usar más procesos que núcleos no siempre acelera el programa. Para trabajo intensivo de CPU, comienza con un número cercano a los núcleos disponibles. Deja margen si el equipo debe seguir atendiendo otras tareas.
Procesar resultados a medida que llegan
imap_unordered() entrega resultados sin esperar el orden original:
from multiprocessing import Pool
def pesado(numero):
return numero, numero ** 3
if __name__ == "__main__":
with Pool(processes=4) as pool:
for resultado in pool.imap_unordered(pesado, range(20)):
print(resultado)Esto reduce la espera cuando las tareas tienen duraciones diferentes y el orden no importa.
El tamaño de chunks
Enviar cada elemento individualmente aumenta la comunicación. map() e imap() aceptan chunksize:
resultados = pool.map(calcular, numeros, chunksize=100)Un bloque mayor reduce sobrecarga, pero puede distribuir peor trabajos desiguales. Prueba con datos representativos.
Los datos deben poder serializarse
Los argumentos y resultados suelen pasar por pickle. Funciones locales, lambdas y algunos objetos abiertos no pueden serializarse fácilmente:
# Mejor: función en el nivel superior del módulo
def transformar(registro):
return registro["valor"] * 2Evita enviar conexiones de base de datos, archivos abiertos o objetos enormes. Crea los recursos necesarios dentro del proceso trabajador.
Memoria separada
Modificar una lista dentro de un proceso no modifica automáticamente la lista del proceso principal:
from multiprocessing import Process
def agregar(lista):
lista.append(99)
if __name__ == "__main__":
valores = [1, 2, 3]
p = Process(target=agregar, args=(valores,))
p.start()
p.join()
print(valores) # sigue igualEsta separación reduce ciertas condiciones de carrera, pero exige mecanismos explícitos para compartir resultados.
Comunicación con Queue
from multiprocessing import Process, Queue
def trabajador(cola, numero):
cola.put(numero ** 2)
if __name__ == "__main__":
cola = Queue()
procesos = [
Process(target=trabajador, args=(cola, n))
for n in range(4)
]
for p in procesos:
p.start()
resultados = [cola.get() for _ in procesos]
for p in procesos:
p.join()
print(resultados)La cola gestiona la sincronización. No uses una lista normal esperando que sea compartida.
Compartir valores pequeños
Value y Array permiten memoria compartida para tipos simples:
from multiprocessing import Process, Value
def incrementar(contador):
for _ in range(1000):
with contador.get_lock():
contador.value += 1
if __name__ == "__main__":
contador = Value("i", 0)
procesos = [Process(target=incrementar, args=(contador,)) for _ in range(4)]
for p in procesos:
p.start()
for p in procesos:
p.join()
print(contador.value)El bloqueo evita actualizaciones perdidas. La memoria compartida debe ser pequeña y cuidadosamente sincronizada.
Manejar excepciones con ProcessPoolExecutor
La interfaz de concurrent.futures facilita obtener resultados y excepciones:
from concurrent.futures import ProcessPoolExecutor, as_completed
def dividir(valor):
return 100 / valor
if __name__ == "__main__":
with ProcessPoolExecutor(max_workers=4) as executor:
futuros = [executor.submit(dividir, n) for n in [5, 2, 0, 4]]
for futuro in as_completed(futuros):
try:
print(futuro.result())
except ZeroDivisionError as error:
print(f"Tarea fallida: {error}")Si nunca llamas a result(), puedes pasar por alto una excepción del trabajador.
Comparar secuencial y paralelo
Un benchmark debe incluir el coste total de crear el pool y transferir datos:
from multiprocessing import Pool
from time import perf_counter
def tarea(numero):
return sum(i * i for i in range(numero))
if __name__ == "__main__":
entradas = [2_000_000] * 8
inicio = perf_counter()
secuencial = [tarea(n) for n in entradas]
print("Secuencial:", perf_counter() - inicio)
inicio = perf_counter()
with Pool() as pool:
paralelo = pool.map(tarea, entradas)
print("Paralelo:", perf_counter() - inicio)Para tareas pequeñas, la versión paralela puede ser más lenta. Multiprocessing funciona cuando el cálculo por tarea compensa la sobrecarga.
Evitar copiar datos gigantes
Enviar un DataFrame o una matriz enorme a cada trabajador puede duplicar memoria y anular la mejora. Divide archivos por rutas, utiliza lotes, memoria compartida especializada o bibliotecas numéricas que ya paralelizan internamente. No cargues todos los datos antes de repartirlos si puedes procesarlos por partes.
Finalizar correctamente
Usa gestores de contexto para pools y executors. Con procesos manuales, llama a join(). Si un trabajador queda bloqueado, investiga la causa antes de utilizar terminate(), porque una terminación abrupta puede dejar archivos o recursos en estado inconsistente.
La documentación oficial de multiprocessing explica procesos, pools, colas y memoria compartida. La documentación de ProcessPoolExecutor presenta una interfaz de alto nivel para tareas y resultados.
Errores frecuentes
Los errores más comunes son omitir el bloque __main__, enviar lambdas o funciones locales, crear demasiados procesos, compartir grandes objetos por serialización, olvidar recoger excepciones y usar paralelismo para tareas demasiado pequeñas. También debes considerar que notebooks y entornos interactivos pueden manejar los procesos de forma distinta a un script normal.
Conclusión
multiprocessing permite paralelismo real para trabajo intensivo de CPU, pero requiere dividir las tareas y limitar la comunicación. Usa funciones serializables, procesa lotes razonables, recoge resultados y mide el tiempo completo. Cuando la carga es principalmente espera, threading o asyncio suelen ser mejores; cuando el cálculo domina y puede separarse, varios procesos pueden reducir significativamente la duración.
