Dask用CPU worker处理数据后调用GPU worker训练XGBoost报序列化错误

你触发报错的直接原因是将不可序列化的Dask Client对象作为参数传入client.submit()。Client对象本身持有socket连接用于和调度器通信，Python的pickle机制无法序列化socket对象，因此抛出对应错误。
同时你对DaskDMatrix的资源指定方式不符合Dask-XGBoost的设计逻辑：DaskDMatrix是分布式感知的构造器，本身需要运行在客户端侧协调各worker的数据分片，不需要通过client.submit提交到worker节点执行。

按以下步骤调整代码即可：

1. 恢复DaskDMatrix的构造逻辑，不要套client.submit

Xy = dxgb.DaskDMatrix(client, X_train, y_train)

2. 在XGBoost训练调用时指定GPU资源要求，调度器会自动将训练任务调度到有GPU资源的worker执行，同时记得在参数中开启GPU训练模式：

params = {
    "tree_method": "gpu_hist", # 必须指定，开启XGBoost的GPU训练路径
    # 其余你的训练参数比如learning_rate、max_depth等放在这里
}
output = dxgb.train(
    client,
    params=params,
    dtrain=Xy,
    num_boost_round=100,
    resources={"GPU": 1} # 在此处指定训练任务需要GPU资源
)

3. （可选）如果你需要严格区分CPU预处理和GPU训练的调度范围，避免预处理任务占用GPU worker资源，可以在CPU阶段的持久化操作中指定仅使用CPU资源：

# parquet读取完成后的持久化操作指定CPU资源
df = dd.read_parquet(
    "hdfs://address/to/the/*.parquet",
    storage_options = {
        "user":user,
        "kerb_ticket":kerb_ticket},
    engine='pyarrow'
).persist(resources={"CPU": 1})

4. （集群配置验证）确保你启动GPU worker时已经正确配置了资源标签，启动命令示例：

dask worker <scheduler_address> --resources "GPU=1"

如果是通过代码启动集群，也要给GPU worker对应的资源配置。

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