[Python] Duplicate values being given to function by Dask distributed.client.submit

I am attemping to use Dask to speed up a function that I was previously running with Pandas.

I have a 3 GB CSV file with columns for ['id','object_names','file_paths','size','upload'] with dtypes {'id':'int','object_names':'str','file_paths':'str','size':'int','upload':'bool'}.

I do:

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)


To read in the data (I tried using a Dask dataframe, but had issues getting literal_eval to work).

Then:

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 searches for an id and list of object_names within a second dataframe, current_objects. If both match, the zip file is already on a backup server, so it returns False, if the id doesn't exist, or it does, but the object_names don't match, it returns True.

Then I do:

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


The problem I'm having is the id values the function sees are not unique.

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


Output:

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


I see no errors. If I add a print(id) before the call to client.submit I see numbers 1 to ~80 sequentially by the time I see backup_2.zip.

Questions:

• Is it possible the function is failing without throwing an error and Dask is retrying?
• Am I using client.submit incorrectly? need to zip and give *args?
• Something else?
• another possible clue is this is currently very slow - zip_list[zip_list.id == id]['object_names'].values[0] is the slowest line shown above, and the function contains two calls to pandas.DataFrame.isin, operating on current_objects, which I previously client.scatter, but if I'm truly giving one line of the large dataframe to each worker, I don't see why it's slow.

Continue reading...