Creación de diferentes tipos de trabajadores que son accedidos utilizando un solo cliente.

12 Jul, 2023 Programación 0

EDITAR: Mi pregunta fue horriblemente formulada, así que la borro y la reformulo por completo aquí.
Daré un resumen:
Estoy tratando de asignar cada cálculo a un trabajador designado que se adapte al tipo de cálculo.
En resumen:
Estoy tratando de ejecutar una simulación, por lo que la represento utilizando una clase de la siguiente forma:

Clase Simulación:
def init (primer_Cliente: Cliente, segundo_Cliente: Cliente)
self.primer_cliente = primer_cliente
self.segundo_cliente = segundo_cliente
def primera_calculacion(entrada):
with primer_cliente.como_actual():
return salida
def segunda_calculacion(entrada):
with segundo_cliente.como_actual():
return salida
def ejecutar(entrada):
return segunda_calculacion(primera_calculacion(entrada))

Este formato tiene desventajas como el hecho de que este objeto de simulación no es pickleable.
Podría editar el objeto Simulación para que contenga solo direcciones y no clientes, por ejemplo, pero siento que debe haber una mejor solución. Por ejemplo, me gustaría que el objeto de simulación funcione de la siguiente manera:

Clase Simulación:
def primera_calculacion(entrada):
cliente = dask.distributed.obtener_cliente()
with cliente.como_actual():
return salida
…

La cuestión es que los trabajadores de Dask que mejor se adaptan para el primer cálculo, son diferentes a los trabajadores de Dask que mejor se adaptan para el segundo cálculo, que es la razón por la que mi objeto Simulación tiene dos clientes que se conectan a dos programadores de tareas diferentes desde el principio. ¿Hay alguna manera de hacer que solo haya un cliente pero dos tipos de programadores y hacer que el cliente sepa ejecutar el primer cálculo en el primer programador y el segundo cálculo en el segundo?

davy.ai

July 12, 2023 at 4:08 pm

Sí, existe una manera de asignar cada cálculo a un trabajador designado que se ajuste al tipo de cálculo utilizando dask y dask-distributed. Una solución es usar etiquetas de dask, que te permiten especificar qué trabajadores deben ejecutar qué tareas en función de sus etiquetas.

Primero, puedes crear dos grupos de trabajadores, uno para cada tipo de cálculo, configurando dos programadores dask con etiquetas diferentes. Por ejemplo:

from dask.distributed import LocalCluster, Client

# Crea los dos grupos con diferentes etiquetas
cluster_first = LocalCluster(n_workers=2, threads_per_worker=1, dashboard_address=None,
                             memory_limit='1GB', silence_logs='error',
                             worker_kwargs={'nthreads': 1},
                             scheduler_kwargs={'dashboard_address': None, 'services': {'bokeh': 8787},
                                               'resources': {'gpu': 0}, 'work_stealing': False,
                                               'default_worker_ttl': '3600s'},
                            )
cluster_first.scheduler.add_resources(my_label='first')

cluster_second = LocalCluster(n_workers=2, threads_per_worker=1, dashboard_address=None,
                             memory_limit='1GB', silence_logs='error',
                             worker_kwargs={'nthreads': 1},
                             scheduler_kwargs={'dashboard_address': None, 'services': {'bokeh': 8788},
                                               'resources': {'gpu': 0}, 'work_stealing': False,
                                               'default_worker_ttl': '3600s'},
                            )
cluster_second.scheduler.add_resources(my_label='second')

# Conéctate a ambos grupos con una única instancia del cliente dask
client = Client([cluster_first.scheduler_address, cluster_second.scheduler_address])

Ahora, en tu clase Simulation, puedes especificar la etiqueta de los trabajadores que deben ejecutar cada cálculo pasando el argumento resources al método client.submit. Por ejemplo:

class Simulation:
    def __init__(self):
        pass

    def first_calculation(self, input):
        with self.client.as_current():
            return output

    def second_calculation(self, input):
        with self.client.as_current():
            return output

    def run(input):
        # Envía el primer cálculo a los trabajadores con la etiqueta "first"
        future1 = self.client.submit(self.first_calculation, input, resources={'my_label': 'first'})
        # Envía el segundo cálculo a los trabajadores con la etiqueta "second"
        future2 = self.client.submit(self.second_calculation, future1, resources={'my_label': 'second'})
        return future2.result()

Al usar etiquetas de dask y enviar tareas con el argumento resources, puedes asignar cada cálculo a un trabajador designado que se ajuste al tipo de cálculo.

es.davy.ai

¿Tienes una pregunta?

Creación de diferentes tipos de trabajadores que son accedidos utilizando un solo cliente.

Answer

davy.ai