Doppelte Werte werden von Dask verteilt.client.submit an die Funktion übergeben ⇐ Python

[Guest]

Ich versuche, Dask zu verwenden, um eine Funktion zu beschleunigen, die ich zuvor mit Pandas ausgeführt habe.
Ich habe eine 3 GB große CSV-Datei mit Spalten für ['id','object_names ','file_paths','size','upload'] mit dtypes {'id':'int','object_names':'str','file_paths':'str','size':'int','upload':'bool'}.
Ich mache:

Code: Select all

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)

Um die Daten einzulesen (ich habe versucht, einen Dask-Datenrahmen zu verwenden, hatte aber Probleme, literal_eval zum Laufen zu bringen).
Dann:

Code: Select all

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))

Code: Select all

compare_zip_list

sucht nach einer ID und einer Liste von Objektnamen innerhalb eines zweiten Datenrahmens, current_objects. Wenn beides übereinstimmt, befindet sich die ZIP-Datei bereits auf einem Backup-Server und gibt daher „Falsch“ zurück. Wenn die ID nicht existiert oder existiert, aber die Objektnamen nicht übereinstimmen, wird „True“ zurückgegeben .
Dann mache ich:

Code: Select all

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

Das Problem, das ich habe, ist, dass die ID-Werte, die die Funktion sieht, nicht eindeutig sind.

Code: Select all

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

Ausgabe:

Code: Select all

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.
...

Ich sehe keine Fehler. Wenn ich vor dem Aufruf von client.submit eine print(id) hinzufüge, sehe ich nacheinander die Nummern 1 bis ~80, bis ich „backup_2.zip“ sehe.
Fragen:< /p>

• Ist es möglich, dass die Funktion fehlschlägt, ohne einen Fehler auszulösen, und Dask es erneut versucht?
• Verwende ich client.submit falsch? Müssen Sie komprimieren und *args angeben?
• Etwas anderes?
• Ein weiterer möglicher Hinweis ist, dass dies derzeit sehr langsam ist - zip_list[zip_list.id == id]['object_names'].values[0] ist die langsamste oben gezeigte Zeile, und die Funktion enthält zwei Aufrufe von pandas.DataFrame.isin, die ausgeführt werden current_objects, die ich zuvor client.scatter, aber wenn ich wirklich jedem Worker eine Zeile des großen Datenrahmens gebe, verstehe ich nicht, warum es langsam ist.