Dask分布式任务等待子任务：可行性、优化与报错排查

1. 嵌套使用Dask的方式是否可行？ #

可行，但不推荐。你当前的写法是在Worker进程里调用client.submit提交子任务，再用wait+result阻塞当前Worker等待子任务完成——这种嵌套会占用Worker资源，导致Worker无法处理其他任务，还容易引发调度器层面的任务调度冲突，也就是你现在遇到的报错场景。

2. 改动量小的优化方案 #

核心思路是避免在Worker里手动提交子任务并阻塞等待，让Dask自动管理任务依赖，同时保留失败捕获能力：

方案一：用delayed构建任务链 #

修改zip_and_upload为纯依赖定义，不在Worker内操作Client，外层直接提交完整任务链：

from dask.distributed import delayed

def zip_and_upload(id, path, file_list):
    # 直接定义任务依赖，不手动提交子任务
    zip_data, name_list = delayed(zip_file_list)(id, file_list, compression)
    upload_task = delayed(upload)(id, zip_data, name_list)
    return upload_task, f'{path}_{id}.zip'

# 外层将delayed对象转为Future提交
zip_and_upload_futures = [client.submit(delayed(zip_and_upload), id, path, file_list) for id, file_list in enumerate(file_lists)]

外层结果处理逻辑可保留，调整异常捕获方式即可：

for future in as_completed(zip_and_upload_futures):
    try:
        upload_fut, zip_name = future.result()
        upload_futures.append(upload_fut)
        print(f'Zip {zip_name} created and added to upload queue.')
    except Exception as e:
        failed_list.append((e, future.traceback()))
    del future

for future in as_completed(upload_futures):
    try:
        result = future.result()
        print(f'Zip {result[0]} uploaded.')
    except Exception as e:
        failed_list.append((e, future.traceback()))
    del future

方案二：将任务拆分到外层管理 #

直接在主进程提交zip_file_list任务，完成后再提交upload任务，完全去掉嵌套逻辑：

# 外层直接提交压缩任务
zip_futures = [client.submit(zip_file_list, id, file_list, compression) for id, file_list in enumerate(file_lists)]
upload_futures = []
failed_list = []

for zip_fut in as_completed(zip_futures):
    try:
        zip_data, name_list = zip_fut.result()
        task_id = zip_fut.key.split('-')[1]  # 从任务key中提取原id
        upload_fut = client.submit(upload, task_id, zip_data, name_list)
        upload_futures.append(upload_fut)
        print(f'Zip {path}_{task_id}.zip created and added to upload queue.')
    except Exception as e:
        failed_list.append((e, zip_fut.traceback()))
    del zip_fut

# 处理上传任务逻辑
for upload_fut in as_completed(upload_futures):
    try:
        result = upload_fut.result()
        print(f'Zip {result[0]} uploaded.')
    except Exception as e:
        failed_list.append((e, upload_fut.traceback()))
    del upload_fut

3. 为何会出现zip_file_list任务queued的报错？ #

原因是Worker资源阻塞导致的调度死锁：
你在zip_and_upload任务中，让Worker进程调用wait(zip_future)+result()，该Worker会被完全阻塞，无法处理任何其他任务。如果调度器刚好把zip_file_list任务分配给了这个已阻塞的Worker，那么zip_file_list就会一直处于queued状态——因为Worker被当前的zip_and_upload任务占用，根本无法执行子任务。

调度器尝试收集zip_file_list任务结果时，发现任务长时间处于排队状态，就会抛出Couldn't gather keys的错误。你期望的“阻塞执行”是让zip_file_list完成后再执行upload，但这种嵌套阻塞的方式反而导致任务无法被调度执行。

内容的提问来源于stack exchange，提问作者Dave