Я новичок в DASK и хотел бы проверить работу DASK в кластере. Кластер имеет головной сервер и несколько других узлов. Я могу войти в другие узлы с помощью простого SSH без пароля, как только я войду в головной сервер.
Я хотел бы запустить простую функцию для перебора большого массива. Функция определяется ниже. Он должен конвертировать dt64 в простой объект datetime.
import xarray as xr
import numpy as np
from dask import compute, delayed
import dask.multiprocessing
from datetime import datetime, timedelta
def converdt64(dt64):
ts = (dt64 - np.datetime64('1970-01-01T00:00:00Z')) / np.timedelta64(1, 's')
return datetime.utcfromtimestamp(ts)
Затем в терминале я выполняю итерацию по массиву 1D с размером N, применяя эту функцию.
values = [delayed(convertdt64)(x) for x in arraydata]
results1 = compute(*values,scheduler='processes’)
При этом используются некоторые ядра на головном сервере, и он работает, хотя и медленно. Затем я попытался запустить функцию на нескольких узлах кластера с помощью клиента, как показано ниже:
from dask.distributed import Client
client = Client("10.140.251.254:8786 »)
results = compute(*values, scheduler='distributed’)
Это не работает вообще. Есть несколько предупреждений и одно сообщение об ошибке, как показано ниже.
distributed.comm.tcp - WARNING - Could not set timeout on TCP stream: [Errno 92] Protocol not available
distributed.comm.tcp - WARNING - Closing dangling stream in <TCP local=tcp://10.140.251.254:57257 remote=tcp://10.140.251.254:8786>
CancelledError: convertdt64-0205ad5e-214b-4683-b5c4-b6a2a6d8e52f
Я также попробовал dask.bag и получил то же сообщение об ошибке. Какие могут быть причины, по которым параллельные вычисления на кластере не работают? Это связано с какой-то конфигурацией сервера / сети или из-за неправильного использования клиента DASK? Заранее спасибо за помощь!
С наилучшими пожеланиями
Шеннон Х