Esta es la primera de una serie de dos partes sobre informática distribuida utilizando Ray. Esta parte muestra cómo usar Ray en su PC local y la parte 2 muestra cómo escalar Ray a clústeres de múltiples servidores en la nube.
Tienes una nueva computadora portátil o de escritorio de 16 núcleos y estás ansioso por probar su potencia con algunos cálculos pesados.
Eres un programador de Python, aunque aún no eres un experto, así que abres tu LLM favorito y le preguntas algo como esto.
"Me gustaría contar el número de números primos dentro de un rango de entrada determinado. Por favor, dame algo de código Python para esto".
Después de unos segundos, el LLM le proporciona un código. Podrías modificarlo un poco mediante un breve intercambio y eventualmente terminarías con algo como esto:
importar matemáticas, tiempo, os def is_prime(n: int) -> bool: si n < 2: devolver Falso si n == 2: devolver Verdadero si n % 2 == 0: devolver Falso r = int(math.isqrt(n)) + 1 para i en rango(3, r, 2): si n % i == 0: devolver Falso devolver Verdadero def count_primes(a: int, b: int) -> int: c = 0 para n en rango(a, b): if is_prime(n): c += 1 return c if __name__ == "__main__": A, B = 10_000_000, 20_000_000 total_cpus = os.cpu_count() o 1 # Iniciar "chunky"; podemos barrer estos fragmentos posteriores = max(4, total_cpus * 2) paso = (B – A) // fragmentos print(f"CPUs~{total_cpus}, fragmentos={fragmentos}") t0 = time.time() resultados =[]para i en rango(fragmentos): s = A + i * paso e = s + paso si i < fragmentos – 1 else B results.append(count_primes(s, e)) total = suma(resultados) print(f"total={total}, time={time.time() – t0:.2f}s")
Ejecutas el programa y funciona perfectamente. El único problema es que lleva bastante tiempo ejecutarlo, tal vez de treinta a sesenta segundos, dependiendo del tamaño del rango de entrada. Probablemente eso sea inaceptable.
¿Qué haces ahora? Tienes varias opciones, siendo probablemente las tres más comunes:
– Paralelizar el código mediante subprocesos o multiprocesamiento.
– Reescribir el código en un lenguaje “rápido” como C o Rust
– Pruebe una biblioteca como Cython, Numba o NumPy
Todas estas son opciones viables, pero cada una tiene desventajas. Las opciones 1 y 3 aumentan significativamente la complejidad de su código y la opción intermedia puede requerir que aprenda un nuevo lenguaje de programación.
¿Y si te dijera que hay otra manera? Uno en el que los cambios necesarios en su código existente se mantendrían al mínimo absoluto. Uno en el que su tiempo de ejecución se distribuye automáticamente entre todos los núcleos disponibles.
Eso es precisamente lo que promete hacer la biblioteca Ray de terceros.
¿Qué es Rayo?
La biblioteca Ray Python es un marco informático distribuido de código abierto diseñado para facilitar la escala de programas Python desde una computadora portátil a un clúster con cambios mínimos de código.
Ray simplifica el escalado y la distribución de cargas de trabajo de aplicaciones con uso intensivo de cómputo (desde aprendizaje profundo hasta procesamiento de datos) entre grupos de computadoras remotas, al tiempo que ofrece mejoras prácticas en el tiempo de ejecución de aplicaciones en su computadora portátil, computadora de escritorio o incluso en un grupo de cómputo remoto basado en la nube.
Ray proporciona un amplio conjunto de bibliotecas e integraciones basadas en un marco de ejecución distribuida flexible, lo que hace que la informática distribuida sea fácil y accesible para todos.
En resumen, Ray le permite paralelizar y distribuir su código Python con un mínimo esfuerzo, ya sea ejecutándolo localmente en una computadora portátil o en un clúster gigante basado en la nube.
Usando rayo
En el resto de este artículo, le explicaré los conceptos básicos del uso de Ray para acelerar el código Python que requiere un uso intensivo de la CPU y configuraremos algunos fragmentos de código de ejemplo para mostrarle lo fácil que es incorporar el poder de Ray en sus propias cargas de trabajo.
Para aprovechar al máximo el uso de Ray, si es un científico de datos o un ingeniero de aprendizaje automático, primero debe comprender algunos conceptos clave. Ray se compone de varios componentes.
Ray Data es una biblioteca escalable diseñada para el procesamiento de datos en tareas de ML e IA. Ofrece API flexibles y de alto rendimiento para tareas de IA, incluida la inferencia por lotes, el preprocesamiento de datos y la ingesta de datos para el entrenamiento de aprendizaje automático.
Ray Train es una biblioteca flexible y escalable diseñada para la capacitación y el ajuste del aprendizaje automático distribuido.
Ray Tune se utiliza para el ajuste de hiperparámetros.
Ray Serve es una biblioteca escalable para implementar modelos para facilitar las API de inferencia en línea.
Ray RLlib se utiliza para el aprendizaje por refuerzo escalable
Como puede ver, Ray está muy centrado en modelos de lenguaje grandes y aplicaciones de IA, pero hay un último componente importante que no he mencionado todavía y es el que usaré en este artículo.
Ray Core está diseñado para escalar aplicaciones Python de uso general con uso intensivo de CPU. Está diseñado para distribuir su carga de trabajo de Python entre todos los núcleos disponibles en cualquier sistema en el que lo esté ejecutando.
Este artículo hablará exclusivamente sobre Ray Core.
Dos conceptos esenciales a comprender dentro de Ray Core son las tareas y los actores.
Las tareas son trabajadores o servicios sin estado implementados utilizando Ray decorando funciones regulares de Python.
Los actores (o trabajadores con estado) se utilizan, por ejemplo, cuando necesita realizar un seguimiento y mantener el estado de las variables dependientes en su clúster distribuido. Los actores se implementan decorando clases regulares de Python.
Tanto los actores como las tareas se definen utilizando el mismo decorador @ray.remote. Una vez definidas, estas tareas se ejecutan con el método especial .remote() proporcionado por Ray. A continuación veremos un ejemplo de esto.
Configurar un entorno de desarrollo
Antes de comenzar a codificar, debemos configurar un entorno de desarrollo para mantener nuestros proyectos aislados para que no interfieran entre sí. Usaré conda para esto, pero siéntete libre de usar la herramienta que prefieras. Ejecutaré mi código usando una computadora portátil Jupyter en un shell WSL2 de Ubuntu en Windows.
$ conda crear -n ray-test python=3.13 -y $ conda activar ray-test (prueba de rayos) $ conda instalar ray [predeterminado]
Ejemplo de código: contar números primos
Volvamos al ejemplo que di al principio: contar el número de números primos dentro del intervalo de 10.000.000 a 20.000.000.
Ejecutaremos nuestro código Python original y calcularemos cuánto tiempo lleva.
importar matemáticas, tiempo, os def is_prime(n: int) -> bool: si n < 2: devolver Falso si n == 2: devolver Verdadero si n % 2 == 0: devolver Falso r = int(math.isqrt(n)) + 1 para i en rango(3, r, 2): si n % i == 0: devolver Falso devolver Verdadero def count_primes(a: int, b: int) -> int: c = 0 para n en rango(a, b): if is_prime(n): c += 1 return c if __name__ == "__main__": A, B = 10_000_000, 20_000_000 total_cpus = os.cpu_count() o 1 # Iniciar "chunky"; podemos barrer estos fragmentos posteriores = max(4, total_cpus * 2) paso = (B – A) // fragmentos print(f"CPUs~{total_cpus}, fragmentos={fragmentos}") t0 = time.time() resultados =[]para i en rango(fragmentos): s = A + i * paso e = s + paso si i < fragmentos – 1 else B results.append(count_primes(s, e)) total = suma(resultados) print(f"total={total}, time={time.time() – t0:.2f}s")
¿Y la salida?
CPU ~ 32, fragmentos = 64 en total = 606028, tiempo = 31,17 s
Ahora bien, ¿podemos mejorar eso usando Ray? Sí, siguiendo este sencillo proceso de 4 pasos.
Paso 1: inicializa Ray. Agregue estas dos líneas al comienzo de su código.
importar rayo ray.init()
Paso 2: crea nuestra función remota. Eso es fácil. Simplemente decora la función que queremos optimizar con el decorador @ray.remote. La función a decorar es la que más trabajo está realizando. En nuestro ejemplo, esa es la función count_primes.
@ray.remote(num_cpus=1) def count_primes(inicio: int, fin: int) -> int: … …
Paso 3: inicie las tareas paralelas. Llame a su función remota usando la directiva .remote Ray.
refs.append(count_primes.remote(s, e))
Paso 4: espere a que se completen todas nuestras tareas. Cada tarea en Ray devuelve un ObjectRef cuando se llama. Esta es una promesa de Ray. Significa que Ray ha iniciado la tarea para que se ejecute de forma remota y Ray devolverá su valor en algún momento en el futuro. Monitorizamos todos los ObjectRefs devueltos al ejecutar tareas utilizando la función ray.get(). Esto se bloquea hasta que se hayan completado todas las tareas.
resultados = ray.get(tareas)
Juntemos todo esto. Como verá, los cambios en nuestro código original son mínimos: solo se agregaron cuatro líneas de código y una declaración impresa para mostrar la cantidad de nodos y núcleos en los que estamos ejecutando.
import math import time # —————————————– # Cambio No. 1 # —————————————– import ray ray.init(auto) def is_prime(n: int) -> bool: if n < 2: return False if n == 2: return True if n % 2 == 0: return False r = int(math.isqrt(n)) + 1 for i in range(3, r, 2): if n % i == 0: return False return True # —————————————– # Cambio No. 2 # —————————————– @ray.remote(num_cpus=1) # bucle de Python puro → 1 CPU por tarea def count_primes(a: int, b: int) -> int: c = 0 for n in range(a, b): if is_prime(n): c += 1 return c if __name__ == "__main__": A, B = 10_000_000, 60_000_000 total_cpus = int(ray.cluster_resources().get("CPU", 1)) # Iniciar "chunky"; podemos barrer estos fragmentos posteriores = max(4, total_cpus * 2) step = (B – A) // fragmentos print(f"nodes={len(ray.nodes())}, CPUs~{total_cpus}, fragmentos={chunks}") t0 = time.time() refs =[]para i en rango(fragmentos): s = A + i * paso e = s + paso si i < fragmentos – 1 else B # —————————————– # Cambio No. 3 # —————————————– refs.append(count_primes.remote(s, e)) # —————————————– # Cambio No. 4 # —————————————– total = sum(ray.get(refs)) print(f"total={total}, tiempo={tiempo.tiempo() – t0:.2f}s")
Ahora bien, ¿ha valido la pena todo esto? Ejecutemos el nuevo código y veamos qué obtenemos.
2025-11-01 13:36:30,650 INFORMACIÓN trabajador.py:2004: inició una instancia de Ray local. Vea el panel en 127.0.0.1:8265 /home/tom/.local/lib/python3.10/site-packages/ray/_private/worker.py:2052: FutureWarning: Consejo: En versiones futuras de Ray, Ray ya no anulará la var env de los dispositivos visibles del acelerador si num_gpus=0 o num_gpus=None (predeterminado). Para habilitar este comportamiento y desactivar este mensaje de error, configure RAY_ACCEL_ENV_VAR_OVERRIDE_ON_ZERO=0 advertencias.warn( nodos=1, CPU~32, fragmentos=64 total=606028, tiempo=3,04 s
Bueno, el resultado habla por sí solo. El código Ray Python es 10 veces más rápido que el código Python normal. No está nada mal.
¿De dónde viene este aumento de velocidad? Bueno, Ray puede distribuir su carga de trabajo a todos los núcleos de su sistema. Un núcleo es como una mini CPU. Cuando ejecutamos nuestro código Python original, usaba solo un núcleo. Está bien, pero si tu CPU tiene más de un núcleo, como tienen la mayoría de las PC modernas, entonces estás dejando dinero sobre la mesa, por así decirlo.
En mi caso, la CPU tiene 24 núcleos, por lo que no sorprende que mi código Ray fuera mucho más rápido que el código que no es Ray.
Empleos de Monitoreo en Ray
Otro punto que vale la pena mencionar es que Ray hace que sea muy fácil monitorear la ejecución de trabajos a través de un panel. Observe que en el resultado que recibimos al ejecutar nuestro código de ejemplo de Ray, vimos esto:
… – Inició una instancia de Ray local. Ver el panel en 127.0.0.1:8265
Muestra un enlace URL local porque lo estoy ejecutando en mi escritorio. Si estuviera ejecutando esto en un clúster, la URL apuntaría a una ubicación en el nodo principal del clúster.
Cuando haces clic en el enlace URL proporcionado, deberías ver algo similar a esto:
Desde esta pantalla principal, puede profundizar para monitorear muchos aspectos de sus programas Ray usando los enlaces del menú en la parte superior de la página.
Usando actores de Ray
Anteriormente mencioné que los actores eran una parte integral del procesamiento central de Ray. Los actores se utilizan para coordinar y compartir datos entre tareas de Ray. Por ejemplo, supongamos que desea establecer un límite global para TODAS las tareas en ejecución que deben cumplir. Supongamos que tiene un conjunto de tareas de trabajo, pero desea asegurarse de que solo se puedan ejecutar un máximo de cinco de esas tareas simultáneamente. Aquí hay un código que podría pensar que funcionaría.
importar matemáticas, tiempo, os def is_prime(n: int) -> bool: si n < 2: devolver Falso si n == 2: devolver Verdadero si n % 2 == 0: devolver Falso r = int(math.isqrt(n)) + 1 para i en rango(3, r, 2): si n % i == 0: devolver Falso devolver Verdadero def count_primes(a: int, b: int) -> int: c = 0 para n en rango(a, b): if is_prime(n): c += 1 return c if __name__ == "__main__": A, B = 10_000_000, 20_000_000 total_cpus = os.cpu_count() o 1 # Iniciar "chunky"; podemos barrer estos fragmentos posteriores = max(4, total_cpus * 2) paso = (B – A) // fragmentos print(f"CPUs~{total_cpus}, fragmentos={fragmentos}") t0 = time.time() resultados =[]para i en rango(fragmentos): s = A + i * paso e = s + paso si i < fragmentos – 1 else B results.append(count_primes(s, e)) total = suma(resultados) print(f"total={total}, time={time.time() – t0:.2f}s")
Hemos utilizado una variable global para limitar la cantidad de tareas en ejecución y el código es sintácticamente correcto y se ejecuta sin errores. Desafortunadamente, no obtendrá el resultado que esperaba. Esto se debe a que cada tarea de Ray se ejecuta en su propio espacio de proceso y tiene su propia copia de la variable global. La variable global NO se comparte entre funciones. Entonces, cuando ejecutemos el código anterior, veremos un resultado como este:
Total de llamadas: 200 GLOBAL_QPS previsto: 5,0 Tiempo esperado si realmente está limitado globalmente: ~40,00 s Tiempo real con 'var global' (roto): 3,80 s QPS del clúster observado: ~52,6 (debería haber sido ~5,0)
Para solucionar este problema, utilizamos un actor. Recuerde que un actor es solo una clase de Python decorada con Ray. Aquí está el código con actores.
import time, ray ray.init(ignore_reinit_error=True, log_to_driver=False) # Este es nuestro actor @ray.remote clase GlobalPacer: """Serializar llamadas para que la tasa en todo el clúster sea <= qps.""" def __init__(self, qps: float): self.interval = 1.0 / qps self.next_time = time.time() def adquirir(self): # Esperar dentro del actor hasta que podamos continuar ahora = time.time() if now < self.next_time: time.sleep(self.next_time – now) # Reserva el siguiente espacio; protegerse contra la deriva self.next_time = max(self.next_time + self.interval, time.time()) return True @ray.remote def call_api_with_limit(n_calls: int, pacer): done = 0 for _ in range(n_calls): # Esperar el permiso global ray.get(pacer.acquire.remote()) # fingir llamada API (no hay sueño extra aquí) done += 1 return done if __name__ == "__main__": NUM_WORKERS = 10 CALLS_EACH = 20 GLOBAL_QPS = 5.0 # límite total de llamadas en todo el clúster = NUM_WORKERS * CALLS_EACH esperado_min_time = total_calls / GLOBAL_QPS pacer = GlobalPacer.remote(GLOBAL_QPS) t0 = time.time() ray.get([call_api_with_limit.remote(CALLS_EACH, pacer) for _ in range(NUM_WORKERS)]) dt = time.time() – t0 print(f"Total de llamadas: {total_calls}") print(f"Límite global de QPS: {GLOBAL_QPS}") print(f"Tiempo esperado (si está limitado a {GLOBAL_QPS} QPS): ~{expected_min_time:.2f}s") print(f"Tiempo real con actor: {dt:.2f}s") print(f"QPS del clúster observado: ~{total_calls/dt:.1f}")
Nuestro código limitador está encapsulado en una clase (GlobalPacer) y decorado con ray.remote, lo que significa que se aplica a todas las tareas en ejecución. Podemos ver la diferencia que esto hace en el resultado ejecutando el código actualizado.
Total de llamadas: 200 Límite de QPS global: 5,0 Tiempo esperado (si tiene un límite de 5,0 QPS): ~40,00 s Tiempo real con el actor: 39,86 s QPS del clúster observado: ~5,0
Resumen
Este artículo presentó Ray, un marco Python de código abierto que facilita la escalación de programas con uso intensivo de computación desde un solo núcleo a varios núcleos o incluso un clúster con cambios mínimos de código.
Mencioné brevemente los componentes clave de Ray (Ray Data, Ray Train, Ray Tune, Ray Serve y Ray Core) y enfaticé que Ray Core es ideal para el escalado de CPU de uso general.
Expliqué algunos de los conceptos esenciales en Ray Core, como la introducción de tareas (funciones paralelas sin estado), actores (trabajadores con estado para coordinación y estado compartido) y ObjectRefs (una promesa futura del valor de retorno de una tarea).
Para mostrar las ventajas de usar Ray, comencé con un ejemplo simple que requiere un uso intensivo de la CPU (contar números primos en un rango) y mostré cómo ejecutarlo en un solo núcleo puede ser lento con una implementación ingenua de Python.
En lugar de reescribir el código en otro idioma o utilizar bibliotecas multiprocesamiento complejas, Ray le permite paralelizar la carga de trabajo en sólo cuatro sencillos pasos y unas pocas líneas de código adicionales:
ray.init() para iniciar Ray Decora tus funciones con @ray.remote para convertirlas en tareas paralelas .remote() para iniciar tareas simultáneamente y ray.get() para recopilar resultados de tareas.
Este enfoque redujo el tiempo de ejecución del ejemplo de conteo de primos de ~30 segundos a ~3 segundos en una máquina de 24 núcleos.
También mencioné lo fácil que es monitorear los trabajos en ejecución en Ray usando su panel integrado y mostré cómo acceder a él.
Finalmente, proporcioné un ejemplo del uso de Ray Actor mostrando por qué las variables globales no son adecuadas para coordinar entre tareas, ya que cada trabajador tiene su propio espacio de memoria.
En la segunda parte de esta serie, veremos cómo llevar las cosas a otro nivel al permitir que los trabajos de Ray utilicen aún más potencia de CPU a medida que escalamos a grandes servidores de múltiples nodos en la nube a través de Amazon Web Services.