Повторяющиеся значения, передаваемые функции Daskраспределенный.client.submit ⇐ Python

[Anonymous]

Я пытаюсь использовать Dask для ускорения функции, которую я ранее запускал в Pandas.
У меня есть CSV-файл размером 3 ГБ со столбцами для ['id','object_names ','file_paths','size','upload'] с dtypes {'id':'int','object_names':'str','file_paths':'str','size':'int','upload':'bool'.
Да:

Код: Выделить всё

zip_list = pd.read_csv(list_path).drop('upload',axis=1)
zip_list.object_names = zip_list.object_names.apply(ast.literal_eval)
zip_list.file_paths = zip_list.file_paths.apply(ast.literal_eval)

Чтобы прочитать данные (я пробовал использовать фрейм данных Dask, но возникли проблемы с работой literal_eval).
Затем:

Код: Выделить всё

cmp_futures = []
for id in zip_list['id'].values:
cmp_futures.append(client.submit(
compare_zip_list,
zip_list[zip_list.id == id]['object_names'].values[0], # should be a 1-d list
id, # should be unique
current_objects))

Код: Выделить всё

compare_zip_list

ищет идентификатор и список имен объектов во втором кадре данных, current_objects. Если оба совпадают, zip-файл уже находится на резервном сервере, поэтому он возвращает False, если идентификатор не существует или существует, но имена объектов не совпадают, он возвращает True .
Тогда я делаю:

Код: Выделить всё

zip_list['upload'] = client.gather(cmp_futures)

Проблема в том, что значения id, которые видит функция, не уникальны.

Код: Выделить всё

def compare_zip_list(object_names, id, current_objects):
dask.distributed.print(id) # or built-in print with flush=True
# code to check if the present object_list and id exists as a zip

Вывод:

Код: Выделить всё

0
backup_0.zip already exists and file lists match - skipping.
0
backup_0.zip already exists and file lists match - skipping.
1
backup_1.zip already exists and file lists match - skipping.
2
backup_2.zip already exists and file lists match - skipping.
1
backup_1.zip already exists and file lists match - skipping.
2
backup_2.zip already exists and file lists match - skipping.
...

Ошибок не вижу. Если я добавлю print(id) перед вызовом client.submit, я увижу числа от 1 до ~80 последовательно к тому моменту, когда увижу backup_2.zip.
Вопросы:< /p>

• Возможно ли, что функция завершается с ошибкой, не выдавая ошибку, и Dask повторяет попытку?
• Использую ли я client.submit неправильно? нужно заархивировать и указать *args?
• Что-то еще?
• еще одна возможная подсказка: в настоящее время это очень медленно - zip_list[zip_list.id == id]['object_names'].values[0] — самая медленная строка, показанная выше, и функция содержит два вызова pandas.DataFrame.isin, работающих на current_objects, который я раньше client.scatter, но если я действительно даю одну строку большого фрейма данных каждому работнику, я не понимаю, почему это медленно.

Подробнее здесь: https://stackoverflow.com/questions/793 ... ent-submit