Dask distributed training with custom evaluation function on HDFS parquet file will cause empty DMatrix warning and corresponding errors

## Description

Somehow using custom metrics with different loading methods from the HDFS parquet file will cause different errors.

- `[ worker xxx ] : worker tcp://xxx has an empty DMatrix.`
- `dispatched_train` compute failed
- `Exception: "XGBoostError('[16:14:37] /workspace/rabit/include/rabit/internal/utils.h:86: Allreduce failed')"`
- `Exception: "ValueError('Found array with 0 sample(s) (shape=(0,)) while a minimum of 1 is required.')"`
- `Exception: "ValueError('training data did not have the following fields: a, b, c, d, e, f, g, h, i, j')"`

## Overview

I have tried a couple of different settings and found passing `custom_metric` to `xgb.dask.train` will cause the empty DMatrix warning.

- Passing Dask DataFrame versus Dask Array
- With or without custom_metric
- Loading data in different ways
1. Single parquet file
2. Concatenate same parquet file multiple times
3. Read multiple parquet files
4. Read the same parquet file multiple times
- Use `client.rebalance()` on Dask DataFrame

## Conclusion

- The issue is not related to using Dask DataFrame or Dask Array (will give different error messages but seems both are caused by getting empty data)
- Using built-in metrics can successfully handle distributed data as well as doing early stopping.
- When loading more than a single parquet file will cause an error, and they are differ
- Loading data in different ways results:
1. Single parquet file => Fine
2. Concatenate same parquet file multiple times
- Dask Array will cause `ValueError: Found array with 0 sample(s) (shape=(0,)) while a minimum of 1 is required.`
- Dask DataFrame will cause `ValueError: training data did not have the following fields: a, b, c, d, e, f, g, h, i, j`
3. Read multiple parquet file => `xgboost.core.XGBoostError: [15:41:39] /workspace/rabit/include/rabit/internal/utils.h:86: Allreduce failed`
4. Read the same parquet file multiple times => Sometimes same as 2.; Sometimes same as 3.
- `client.rebalance()` has not effect

## Reproduce The Problem

### Environments

Packages

- XGBoost: 2.0.2
- PyArrow: 14.0.0
- Dask: 2023.5.0
- Scikit Learn: 1.3.2
- Pandas: 2.0.3

System

- Ubuntu 18.04.6 LTS
- Python 3.8.13
- Hadoop 3.3.6

### Settings

Here is a reproducable example:

1. Have a Dask Cluster running
- Scheduler: 192.168.222.236
- Worker: 192.168.222.{235,236,237}
2. Have HDFS running on all three machine same as Dask workers
- Name Node: 192.168.222.236:9000

### 1. Create sample data and store it on HDFS (fixed code)

First, import and create example data

```python
import xgboost as xgb
import pandas as pd
import dask.dataframe as dd
from sklearn.datasets import make_regression
from sklearn.metrics import mean_squared_error
import dask.distributed
import pyarrow
import pyarrow.fs
import pyarrow.parquet
import traceback

DIMENSION = 10
NUM_BOOST_ROUND = 10

client = dask.distributed.Client(address="tcp://192.168.222.236:8786")
# Starting with clean workers
client.restart()

# Creating sample regression data
X_np, y_np = make_regression(n_samples=1000, n_features=DIMENSION)
X_test_np, y_test_np = make_regression(n_samples=300, n_features=DIMENSION)
columns = [chr(i) for i in range(ord('a'), ord('a') + DIMENSION)]
X_df = pd.DataFrame(X_np, columns=columns)
y_df = pd.Series(y_np, name='label').to_frame()
X_test_df = pd.DataFrame(X_test_np, columns=columns)
y_test_df = pd.Series(y_test_np, name='label').to_frame()
```

Save the file using PyArrow (same effect as using `hdfs dfs -put` to upload local parquet to HDFS)

```python
hdfs = pyarrow.fs.HadoopFileSystem(
host='192.168.222.236', port=9000, user='hadoop')

pyarrow.parquet.write_table(pyarrow.Table.from_pandas(pd.concat(
[X_df, y_df], axis=1)), '/user/lidawei/xgboost_debug/train_df.parquet', filesystem=hdfs)

pyarrow.parquet.write_table(pyarrow.Table.from_pandas(pd.concat(
[X_test_df, y_test_df], axis=1)), '/user/lidawei/xgboost_debug/test_df.parquet', filesystem=hdfs)
```

### 2. Different ways of loading training data (with corresponding error message)

#### 1. Loading training data as a single parquet file (Working fine)

```python
df = dd.read_parquet('hdfs://hadoop@192.168.222.236:9000/user/lidawei/xgboost_debug/train_df.parquet')

# or

DUPLICATE = 1
df = dd.read_parquet(['hdfs://hadoop@192.168.222.236:9000/user/lidawei/xgboost_debug/train_df.parquet'] * DUPLICATE)
```

No warning and error

#### 2. Loading training data by concatenating the same parquet file multiple times

```python
df = dd.concat([dd.read_parquet('hdfs://hadoop@192.168.222.236:9000/user/lidawei/xgboost_debug/train_df.parquet') for _ in range(DUPLICATE)])
```

Client-side error log

```text
========== Train Dask Array with Custom Metric ==========
Traceback (most recent call last):
File "xgbooxt_reproduce_HDFS.py", line 145, in
print(xgb.dask.train(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 731, in inner_f
return func(**kwargs)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 1079, in train
return client.sync(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/utils.py", line 351, in sync
return sync(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/utils.py", line 418, in sync
raise exc.with_traceback(tb)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/utils.py", line 391, in f
result = yield future
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/tornado/gen.py", line 767, in run
value = future.result()
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 1015, in _train_async
results = await map_worker_partitions(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 532, in map_worker_partitions
results = await client.gather(futures)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/client.py", line 2224, in _gather
raise exception.with_traceback(traceback)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 986, in dispatched_train
booster = worker_train(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 731, in inner_f
return func(**kwargs)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/training.py", line 182, in train
if cb_container.after_iteration(bst, i, dtrain, evals):
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/callback.py", line 238, in after_iteration
score: str = model.eval_set(evals, epoch, self.metric, self._output_margin)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 2140, in eval_set
feval_ret = feval(
File "xgbooxt_reproduce_HDFS.py", line 140, in rmse
return 'custom_rmse', mean_squared_error(y_true, y_pred) ** 0.5
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/sklearn/utils/_param_validation.py", line 214, in wrapper
return func(*args, **kwargs)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/sklearn/metrics/_regression.py", line 474, in mean_squared_error
y_type, y_true, y_pred, multioutput = _check_reg_targets(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/sklearn/metrics/_regression.py", line 100, in _check_reg_targets
y_true = check_array(y_true, ensure_2d=False, dtype=dtype)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/sklearn/utils/validation.py", line 967, in check_array
raise ValueError(
ValueError: Found array with 0 sample(s) (shape=(0,)) while a minimum of 1 is required.
========== Train Dask DataFrame with Custom Metric==========
Traceback (most recent call last):
File "xgbooxt_reproduce_HDFS.py", line 162, in
print(xgb.dask.train(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 731, in inner_f
return func(**kwargs)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 1079, in train
return client.sync(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/utils.py", line 351, in sync
return sync(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/utils.py", line 418, in sync
raise exc.with_traceback(tb)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/utils.py", line 391, in f
result = yield future
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/tornado/gen.py", line 767, in run
value = future.result()
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 1015, in _train_async
results = await map_worker_partitions(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 532, in map_worker_partitions
results = await client.gather(futures)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/client.py", line 2224, in _gather
raise exception.with_traceback(traceback)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 986, in dispatched_train
booster = worker_train(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 731, in inner_f
return func(**kwargs)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/training.py", line 182, in train
if cb_container.after_iteration(bst, i, dtrain, evals):
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/callback.py", line 238, in after_iteration
score: str = model.eval_set(evals, epoch, self.metric, self._output_margin)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 2141, in eval_set
self.predict(dmat, training=False, output_margin=output_margin),
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 2273, in predict
self._validate_features(fn)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 2943, in _validate_features
raise ValueError(
ValueError: training data did not have the following fields: a, b, c, d, e, f, g, h, i, j
```

Simplified Server-side log

```text
[ worker 192.168.222.236 ] : worker tcp://192.168.222.236:42169 has an empty DMatrix. [ worker 192.168.222.235 ] : [16:05:46] task [xgboost.dask-tcp://192.168.222.235:45949]:tcp://192.168.222.235:45949 got new rank 0 [ worker 192.168.222.235 ] : worker tcp://192.168.222.235:45949 has an empty DMatrix. [ scheduler 192.168.222.236:8786 ] : [0] Train-rmse:129.40915 Valid-rmse:188.03456 [ worker 192.168.222.235 ] : [16:05:46] WARNING: /workspace/src/common/error_msg.cc:52: Empty dataset at worker: 0
[ worker 192.168.222.237 ] : [16:05:48] task [xgboost.dask-tcp://192.168.222.237:34307]:tcp://192.168.222.237:34307 got new rank 1 [ worker 192.168.222.237 ] : worker tcp://192.168.222.237:34307 has an empty DMatrix. [ worker 192.168.222.237 ] : 2023-11-20 16:05:48,904 - distributed.worker - WARNING - Compute Failed [ worker 192.168.222.237 ] : Key: dispatched_train-23671ca1-2cf2-4dff-99e0-899f57e0cb92 [ worker 192.168.222.237 ] : Function: dispatched_train [ worker 192.168.222.237 ] : args: ({'objective': 'reg:squarederror', 'eval_metric': ['rmse'], 'tree_method': 'hist'}, {'DMLC_NUM_WORKER': 2, 'DMLC_TRACKER_URI': '192.168.222.236', 'DMLC_TRACKER_PORT': 53893}, 140004864694304, ['Train', 'Valid'], [140004864694304, 140008368830208], {'feature_names': None, 'feature_types': None, 'feature_weights': None, 'missing': nan, 'enable_categorical': False, 'parts': None, 'is_quantile': False}, {'feature_names': None, 'feature_types': None, 'feature_weights': None, 'missing': nan, 'enable_categorical': False, 'parts': None, 'is_quantile': False}, {'feature_names': None, 'feature_types': None, 'feature_weights': None, 'missing': nan, 'enable_categorical': False, 'parts': [{'data': array([[-2.4454075 , -0.69967897, 0.12554679, ..., -1.6863146 , [ worker 192.168.222.237 ] : 1.06193218, 0.23246532], [ worker 192.168.222.237 ] : [ 0.71199377, 0.29535539, 1.95610334, ..., -1.52806446, [ worker 192.168.222.237 ] : 0.39512471, -0.74503689], [ worker 192.168.222.237 ] : [-0.84755979, -1.71332768, 0.32000036, ..., 0.7957702 , [ worker 192.168.222.237 ] : -1.01396365, -0.92493896], [ worker 192.168.222.237 ] : [ worker 192.168.222.237 ] : kwargs: {} [ worker 192.168.222.237 ] : Exception: "ValueError('Found array with 0 sample(s) (shape=(0,)) while a minimum of 1 is required.')" [ worker 192.168.222.237 ] : [ worker 192.168.222.236 ] : [16:05:48] task [xgboost.dask-tcp://192.168.222.236:42169]:tcp://192.168.222.236:42169 got new rank 0 [ worker 192.168.222.236 ] : worker tcp://192.168.222.236:42169 has an empty DMatrix. [ worker 192.168.222.236 ] : 2023-11-20 16:05:48,828 - distributed.worker - WARNING - Compute Failed [ worker 192.168.222.236 ] : Key: dispatched_train-de962224-c3df-417e-b77e-db0b6db5317c [ worker 192.168.222.236 ] : Function: dispatched_train [ worker 192.168.222.236 ] : args: ({'objective': 'reg:squarederror', 'eval_metric': ['rmse'], 'tree_method': 'hist'}, {'DMLC_NUM_WORKER': 2, 'DMLC_TRACKER_URI': '192.168.222.236', 'DMLC_TRACKER_PORT': 53893}, 140004864694304, ['Train', 'Valid'], [140004864694304, 140008368830208], {'feature_names': None, 'feature_types': None, 'feature_weights': None, 'missing': nan, 'enable_categorical': False, 'parts': [{'data': array([[-0.38422539, -1.99169075, -0.39816466, ..., 0.00458752, [ worker 192.168.222.236 ] : 0.03300869, -0.32428884], [ worker 192.168.222.236 ] : [-0.50862924, -0.06078752, 0.51296917, ..., 1.33159113, [ worker 192.168.222.236 ] : 0.08649411, 1.45877079], [ worker 192.168.222.236 ] : [ 0.73515627, -0.19264822, -1.02095918, ..., -1.38732397, [ worker 192.168.222.236 ] : 0.63085157, -0.74546633],
[ worker 192.168.222.236 ] : ...,
[ worker 192.168.222.236 ] : [-0.71538138, -0.39522918, -0.30154621, ..., -0.88965097,
[ worker 192.168.222.236 ] : -0.78303422, -0.86534836],
[ worker 192.168.222.236 ] : [-0.7535492 , -1.10356109, -0.675427 , ..., -0.46816204,
[ worker 192.168.222.236 ] : 0.64806994, -0.69083135],
[ worker 192.168.222.236 ] : [-0.97014447, 0.5175249 , 0.24838003, ..., 0.49199655,
[ worker 192.168.222.236 ] : 0.60352595, 1.50841363]]), 'l
[ worker 192.168.222.236 ] : kwargs: {}
[ worker 192.168.222.236 ] : Exception: "ValueError('Found array with 0 sample(s) (shape=(0,)) while a minimum of 1 is required.')"
[ worker 192.168.222.235 ] : [16:05:49] task [xgboost.dask-tcp://192.168.222.235:45949]:tcp://192.168.222.235:45949 got new rank 0
[ worker 192.168.222.235 ] : worker tcp://192.168.222.235:45949 has an empty DMatrix.
[ scheduler 192.168.222.236:8786 ] : [9] Train-rmse:26.07020 Valid-rmse:144.78493
[ scheduler 192.168.222.236:8786 ] : 2023-11-20 16:05:48,204 - distributed.worker - INFO - Run out-of-band function '_start_tracker'
[ scheduler 192.168.222.236:8786 ] : 2023-11-20 16:05:48,785 - distributed.worker - INFO - Run out-of-band function '_start_tracker'
[ worker 192.168.222.236 ] : [16:05:49] task [xgboost.dask-tcp://192.168.222.236:42169]:tcp://192.168.222.236:42169 got new rank 1
[ worker 192.168.222.236 ] : worker tcp://192.168.222.236:42169 has an empty DMatrix.
[ worker 192.168.222.236 ] : 2023-11-20 16:05:49,781 - distributed.worker - WARNING - Compute Failed
[ worker 192.168.222.236 ] : Key: dispatched_train-cc3b97a0-113a-4bf0-813f-ff5ac671fc7c
[ worker 192.168.222.236 ] : Function: dispatched_train
[ worker 192.168.222.236 ] : args: ({'objective': 'reg:squarederror', 'eval_metric': ['rmse'], 'tree_method': 'hist'}, {'DMLC_NUM_WORKER': 2, 'DMLC_TRACKER_URI': '192.168.222.236', 'DMLC_TRACKER_PORT': 50529}, 140004907039760, ['Train', 'Valid'], [140004907039760, 140004907040720], {'feature_names': None, 'feature_types': None, 'feature_weights': None, 'missing': nan, 'enable_categorical': False, 'parts': None, 'is_quantile': False}, {'feature_names': None, 'feature_types': None, 'feature_weights': None, 'missing': nan, 'enable_categorical': False, 'parts': None, 'is_quantile': False}, {'feature_names': None, 'feature_types': None, 'feature_weights': None, 'missing': nan, 'enable_categorical': False, 'parts': [{'data': a b c ... h i j
[ worker 192.168.222.236 ] : 0 -2.445408 -0.699679 0.125547 ... -1.686315 1.061932 0.232465
[ worker 192.168.222.236 ] : 1 0.711994 0.295355 1.956103 ... -1.528064 0.395125 -0.745037
[ worker 192.168.222.236 ] : 2 -0.847560 -1.713328 0.320000 ... 0.795770 -1.013964 -0.924939
[ worker 192.168.222.236 ] : 3 -0.847362 1.144842 0.379
[ worker 192.168.222.236 ] : kwargs: {}
[ worker 192.168.222.236 ] : Exception: "ValueError('training data did not have the following fields: a, b, c, d, e, f, g, h, i, j')"
[ worker 192.168.222.236 ] :
[ worker 192.168.222.235 ] : 2023-11-20 16:05:49,911 - distributed.worker - WARNING - Compute Failed
[ worker 192.168.222.235 ] : Key: dispatched_train-27e46683-c75f-472a-b827-3fdbdd718114
[ worker 192.168.222.235 ] : Function: dispatched_train
[ worker 192.168.222.235 ] : args: ({'objective': 'reg:squarederror', 'eval_metric': ['rmse'], 'tree_method': 'hist'}, {'DMLC_NUM_WORKER': 2, 'DMLC_TRACKER_URI': '192.168.222.236', 'DMLC_TRACKER_PORT': 50529}, 140004907039760, ['Train', 'Valid'], [140004907039760, 140004907040720], {'feature_names': None, 'feature_types': None, 'feature_weights': None, 'missing': nan, 'enable_categorical': False, 'parts': [{'data': a b c ... h i j
[ worker 192.168.222.235 ] : 0 -0.384225 -1.991691 -0.398165 ... 0.004588 0.033009 -0.324289
[ worker 192.168.222.235 ] : 1 -0.508629 -0.060788 0.512969 ... 1.331591 0.086494 1.458771
[ worker 192.168.222.235 ] : 2 0.735156 -0.192648 -1.020959 ... -1.387324 0.630852 -0.745466
[ worker 192.168.222.235 ] : 3 -0.069762 1.148025 -0.961715 ... -0.020099 -1.295274 -0.012636
[ worker 192.168.222.235 ] : 4 -1.417914 -0.756072 -0.576239 ... 0.196316 0.230254 -0.970875
[ worker 192.168.222.235 ] : .. ... ... ... ... ... ... ...
[ worker 192.168.222.235 ] : 995 -1.020673 -0.317122 -0.306466 ... -0.649064 0.701003 2.405321
[ worker 192.168.222.235 ] : 996 0.078581 0.224891 0.093955 ... -0.082395 0.635476 0.08
[ worker 192.168.222.235 ] : kwargs: {}
[ worker 192.168.222.235 ] : Exception: "ValueError('training data did not have the following fields: a, b, c, d, e, f, g, h, i, j')"
```

#### 3. Loading training data by reading multiple parquet files

```python
df = dd.read_parquet(['hdfs://hadoop@192.168.222.236:9000/user/lidawei/xgboost_debug/train_df.parquet', 'hdfs://hadoop@192.168.222.236:9000/user/lidawei/xgboost_debug/test_df.parquet'])
```
Client-side error log

```text
========== Train Dask Array with Custom Metric==========
Traceback (most recent call last):
File "xgbooxt_reproduce_HDFS.py", line 145, in
print(xgb.dask.train(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 731, in inner_f
return func(**kwargs)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 1079, in train
return client.sync(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/utils.py", line 351, in sync
return sync(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/utils.py", line 418, in sync
raise exc.with_traceback(tb)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/utils.py", line 391, in f
result = yield future
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/tornado/gen.py", line 767, in run
value = future.result()
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 1015, in _train_async
results = await map_worker_partitions(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 532, in map_worker_partitions
results = await client.gather(futures)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/client.py", line 2224, in _gather
raise exception.with_traceback(traceback)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 986, in dispatched_train
booster = worker_train(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 731, in inner_f
return func(**kwargs)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/training.py", line 182, in train
if cb_container.after_iteration(bst, i, dtrain, evals):
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/callback.py", line 240, in after_iteration
self._update_history(metric_score, epoch)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/callback.py", line 207, in _update_history
x = _allreduce_metric(x)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/callback.py", line 118, in _allreduce_metric
arr = collective.allreduce(arr, collective.Op.SUM) / world
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/collective.py", line 235, in allreduce
_check_call(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 283, in _check_call
raise XGBoostError(py_str(_LIB.XGBGetLastError()))
xgboost.core.XGBoostError: [16:12:02] /workspace/rabit/include/rabit/internal/utils.h:86: Allreduce failed
========== Train Dask DataFrame with Custom Metric==========
Traceback (most recent call last):
File "xgbooxt_reproduce_HDFS.py", line 162, in
print(xgb.dask.train(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 731, in inner_f
return func(**kwargs)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 1079, in train
return client.sync(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/utils.py", line 351, in sync
return sync(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/utils.py", line 418, in sync
raise exc.with_traceback(tb)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/utils.py", line 391, in f
result = yield future
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/tornado/gen.py", line 767, in run
value = future.result()
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 1015, in _train_async
results = await map_worker_partitions(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 532, in map_worker_partitions
results = await client.gather(futures)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/client.py", line 2224, in _gather
raise exception.with_traceback(traceback)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 986, in dispatched_train
booster = worker_train(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 731, in inner_f
return func(**kwargs)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/training.py", line 182, in train
if cb_container.after_iteration(bst, i, dtrain, evals):
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/callback.py", line 240, in after_iteration
self._update_history(metric_score, epoch)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/callback.py", line 207, in _update_history
x = _allreduce_metric(x)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/callback.py", line 118, in _allreduce_metric
arr = collective.allreduce(arr, collective.Op.SUM) / world
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/collective.py", line 235, in allreduce
_check_call(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 283, in _check_call
raise XGBoostError(py_str(_LIB.XGBGetLastError()))
xgboost.core.XGBoostError: [16:12:02] /workspace/rabit/include/rabit/internal/utils.h:86: Allreduce failed
```

Simplified Server-side log

```text
[ worker 192.168.222.236 ] : [16:11:58] task [xgboost.dask-tcp://192.168.222.236:41425]:tcp://192.168.222.236:41425 got new rank 0 [ worker 192.168.222.237 ] : [16:11:58] task [xgboost.dask-tcp://192.168.222.237:43879]:tcp://192.168.222.237:43879 got new rank 1 [ worker 192.168.222.237 ] : worker tcp://192.168.222.237:43879 has an empty DMatrix. [ worker 192.168.222.236 ] : 2023-11-20 16:12:02,047 - distributed.worker - WARNING - Compute Failed [ worker 192.168.222.236 ] : Key: dispatched_train-44f1449b-dba2-44b7-8c4d-0019036d70dd [ worker 192.168.222.236 ] : Function: dispatched_train [ worker 192.168.222.236 ] : args: ({'objective': 'reg:squarederror', 'eval_metric': ['rmse'], 'tree_method': 'hist'}, {'DMLC_NUM_WORKER': 2, 'DMLC_TRACKER_URI': '192.168.222.236', 'DMLC_TRACKER_PORT': 50969}, 140310543314272, ['Train', 'Valid'], [140310543314272, 140310545768208], {'feature_names': None, 'feature_types': None, 'feature_weights': None, 'missing': nan, 'enable_categorical': False, 'parts': [{'data': array([[-1.77974585, -0.45840995, 0.18663852, ..., 0.98678994, [ worker 192.168.222.236 ] : 0.83816918, 0.35383511], [ worker 192.168.222.236 ] : [ 0.80087277, -0.91560538, 0.87568695, ..., 0.85230028, [ worker 192.168.222.236 ] : 0.36662493, 0.68904372], [ worker 192.168.222.236 ] : [ 3.02255112, -2.01013149, 0.57121451, ..., -0.74871302, [ worker 192.168.222.236 ] : -0.16987018, -2.32294019], [ worker 192.168.222.236 ] : ..., [ worker 192.168.222.236 ] : [-0.76749175, -1.55288495, 0.54106476, ..., -0.06773743, [ worker 192.168.222.236 ] : 1.28417965, 0.69102612], [ worker 192.168.222.236 ] : [ 1.52846597, -0.25484983, -0.43779188, ..., 0.7975776 , [ worker 192.168.222.236 ] : 0.09877002, -0.48734085], [ worker 192.168.222.236 ] : [ 0.58914046, 0.04385179, 0.26338388, ..., -0.2069298 , [ worker 192.168.222.236 ] : -0.18323749, 0.52244406]]), 'l [ worker 192.168.222.236 ] : kwargs: {} [ worker 192.168.222.236 ] : Exception: "XGBoostError('[16:12:02] /workspace/rabit/include/rabit/internal/utils.h:86: Allreduce failed')" [ worker 192.168.222.236 ] : [ worker 192.168.222.236 ] : [16:12:02] task [xgboost.dask-tcp://192.168.222.236:41425]:tcp://192.168.222.236:41425 got new rank 0 [ worker 192.168.222.236 ] : 2023-11-20 16:12:02,659 - distributed.worker - WARNING - Compute Failed [ worker 192.168.222.236 ] : Key: dispatched_train-470f45aa-21ff-4b8d-b406-1ae2f6286a58 [ worker 192.168.222.236 ] : Function: dispatched_train [ worker 192.168.222.236 ] : args: ({'objective': 'reg:squarederror', 'eval_metric': ['rmse'], 'tree_method': 'hist'}, {'DMLC_NUM_WORKER': 2, 'DMLC_TRACKER_URI': '192.168.222.236', 'DMLC_TRACKER_PORT': 35465}, 140306915523024, ['Train', 'Valid'], [140306915523024, 140306915545200], {'feature_names': None, 'feature_types': None, 'feature_weights': None, 'missing': nan, 'enable_categorical': False, 'parts': [{'data': a b c ... h i j [ worker 192.168.222.236 ] : 0 -1.779746 -0.458410 0.186639 ... 0.986790 0.838169 0.353835 [ worker 192.168.222.236 ] : 1 0.800873 -0.915605 0.875687 ... 0.852300 0.366625 0.689044 [ worker 192.168.222.236 ] : 2 3.022551 -2.010131 0.571215 ... -0.748713 -0.169870 -2.322940
[ worker 192.168.222.236 ] : 3 0.949807 -0.279788 0.269398 ... -0.914971 -0.376351 -0.661514
[ worker 192.168.222.236 ] : 4 -1.184326 -0.417169 0.590338 ... 0.477586 -0.429590 1.264293
[ worker 192.168.222.236 ] : .. ... ... ... ... ... ... ...
[ worker 192.168.222.236 ] : 295 -0.963161 -0.433303 0.262199 ... 1.092341 0.843093 -0.807826
[ worker 192.168.222.236 ] : 296 0.939847 0.008250 -1.934408 ... 0.496437 1.132208 0.94
[ worker 192.168.222.236 ] : kwargs: {}
[ worker 192.168.222.236 ] : Exception: "XGBoostError('[16:12:02] /workspace/rabit/include/rabit/internal/utils.h:86: Allreduce failed')"
[ worker 192.168.222.237 ] : 2023-11-20 16:12:02,239 - distributed.worker - WARNING - Compute Failed
[ worker 192.168.222.237 ] : Key: dispatched_train-2deeea79-18c9-46d4-9332-4481320da0f6
[ worker 192.168.222.237 ] : Function: dispatched_train
[ worker 192.168.222.237 ] : args: ({'objective': 'reg:squarederror', 'eval_metric': ['rmse'], 'tree_method': 'hist'}, {'DMLC_NUM_WORKER': 2, 'DMLC_TRACKER_URI': '192.168.222.236', 'DMLC_TRACKER_PORT': 50969}, 140310543314272, ['Train', 'Valid'], [140310543314272, 140310545768208], {'feature_names': None, 'feature_types': None, 'feature_weights': None, 'missing': nan, 'enable_categorical': False, 'parts': [{'data': array([[-1.28451619, 1.07712735, 1.34822904, ..., 0.46403871,
[ worker 192.168.222.237 ] : -1.17450263, -0.45740666],
[ worker 192.168.222.237 ] : [ 0.27512656, 1.11584433, 1.14536888, ..., -1.29408342,
[ worker 192.168.222.237 ] : -0.73517287, 0.5221345 ],
[ worker 192.168.222.237 ] : [ 0.0726516 , -0.33515093, 0.88702377, ..., 0.79670704,
[ worker 192.168.222.237 ] : -1.0069391 , 0.52126106],
[ worker 192.168.222.237 ] : ...,
[ worker 192.168.222.237 ] : [-0.11135348, 0.17358024, 0.31739223, ..., 0.91456373,
[ worker 192.168.222.237 ] : -1.88297657, -0.50686043],
[ worker 192.168.222.237 ] : [-0.87700553, -0.88337762, 0.27517537, ..., 0.49615681,
[ worker 192.168.222.237 ] : 0.45352585, -0.06529842],
[ worker 192.168.222.237 ] : [-1.16335698, -0.82854647, -1.43802943, ..., -0.2049112 ,
[ worker 192.168.222.237 ] : 2.95637432, -0.96959806]]), 'l
[ worker 192.168.222.237 ] : kwargs: {}
[ worker 192.168.222.237 ] : Exception: "ValueError('Found array with 0 sample(s) (shape=(0,)) while a minimum of 1 is required.')"
[ worker 192.168.222.237 ] :
[ worker 192.168.222.237 ] : [16:12:02] task [xgboost.dask-tcp://192.168.222.237:43879]:tcp://192.168.222.237:43879 got new rank 1
[ worker 192.168.222.237 ] : worker tcp://192.168.222.237:43879 has an empty DMatrix.
[ worker 192.168.222.237 ] : 2023-11-20 16:12:03,967 - distributed.worker - WARNING - Compute Failed
[ worker 192.168.222.237 ] : Key: dispatched_train-a7675aaa-6675-491f-bea4-9bb2a4e95575
[ worker 192.168.222.237 ] : Function: dispatched_train
[ worker 192.168.222.237 ] : args: ({'objective': 'reg:squarederror', 'eval_metric': ['rmse'], 'tree_method': 'hist'}, {'DMLC_NUM_WORKER': 2, 'DMLC_TRACKER_URI': '192.168.222.236', 'DMLC_TRACKER_PORT': 35465}, 140306915523024, ['Train', 'Valid'], [140306915523024, 140306915545200], {'feature_names': None, 'feature_types': None, 'feature_weights': None, 'missing': nan, 'enable_categorical': False, 'parts': [{'data': a b c ... h i j
[ worker 192.168.222.237 ] : 0 -1.284516 1.077127 1.348229 ... 0.464039 -1.174503 -0.457407
[ worker 192.168.222.237 ] : 1 0.275127 1.115844 1.145369 ... -1.294083 -0.735173 0.522135
[ worker 192.168.222.237 ] : 2 0.072652 -0.335151 0.887024 ... 0.796707 -1.006939 0.521261
[ worker 192.168.222.237 ] : 3 0.321203 -0.535654 0.040780 ... -2.500438 0.800021 -0.372579
[ worker 192.168.222.237 ] : 4 1.481144 -1.172350 -0.391634 ... -0.482940 1.836906 -0.398469
[ worker 192.168.222.237 ] : .. ... ... ... ... ... ... ...
[ worker 192.168.222.237 ] : 995 2.751948 -2.062379 2.190053 ... 0.337114 -0.051003 0.370824
[ worker 192.168.222.237 ] : 996 -1.143378 0.131686 -0.109620 ... 0.459407 -0.005304 -1.53
[ worker 192.168.222.237 ] : kwargs: {}
[ worker 192.168.222.237 ] : Exception: "ValueError('training data did not have the following fields: a, b, c, d, e, f, g, h, i, j')"
```

#### 4. Loading training data by reading the same parquet file multiple times

```python
DUPLICATE = 6
df = dd.read_parquet(['hdfs://hadoop@192.168.222.236:9000/user/lidawei/xgboost_debug/train_df.parquet'] * DUPLICATE)
```
Client-side error log

```text
========== Train Dask Array with Custom Metric==========
Traceback (most recent call last):
File "xgbooxt_reproduce_HDFS.py", line 145, in
print(xgb.dask.train(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 731, in inner_f
return func(**kwargs)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 1079, in train
return client.sync(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/utils.py", line 351, in sync
return sync(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/utils.py", line 418, in sync
raise exc.with_traceback(tb)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/utils.py", line 391, in f
result = yield future
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/tornado/gen.py", line 767, in run
value = future.result()
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 1015, in _train_async
results = await map_worker_partitions(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 532, in map_worker_partitions
results = await client.gather(futures)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/client.py", line 2224, in _gather
raise exception.with_traceback(traceback)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 986, in dispatched_train
booster = worker_train(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 731, in inner_f
return func(**kwargs)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/training.py", line 182, in train
if cb_container.after_iteration(bst, i, dtrain, evals):
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/callback.py", line 240, in after_iteration
self._update_history(metric_score, epoch)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/callback.py", line 207, in _update_history
x = _allreduce_metric(x)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/callback.py", line 118, in _allreduce_metric
arr = collective.allreduce(arr, collective.Op.SUM) / world
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/collective.py", line 235, in allreduce
_check_call(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 283, in _check_call
raise XGBoostError(py_str(_LIB.XGBGetLastError()))
xgboost.core.XGBoostError: [16:14:37] /workspace/rabit/include/rabit/internal/utils.h:86: Allreduce failed
========== Train Dask DataFrame with Custom Metric==========
Traceback (most recent call last):
File "xgbooxt_reproduce_HDFS.py", line 162, in
print(xgb.dask.train(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 731, in inner_f
return func(**kwargs)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 1079, in train
return client.sync(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/utils.py", line 351, in sync
return sync(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/utils.py", line 418, in sync
raise exc.with_traceback(tb)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/utils.py", line 391, in f
result = yield future
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/tornado/gen.py", line 767, in run
value = future.result()
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 1015, in _train_async
results = await map_worker_partitions(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 532, in map_worker_partitions
results = await client.gather(futures)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/distributed/client.py", line 2224, in _gather
raise exception.with_traceback(traceback)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/dask.py", line 986, in dispatched_train
booster = worker_train(
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 731, in inner_f
return func(**kwargs)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/training.py", line 182, in train
if cb_container.after_iteration(bst, i, dtrain, evals):
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/callback.py", line 238, in after_iteration
score: str = model.eval_set(evals, epoch, self.metric, self._output_margin)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 2141, in eval_set
self.predict(dmat, training=False, output_margin=output_margin),
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 2273, in predict
self._validate_features(fn)
File "/mnt/NAS/sda/ShareFolder/lidawei/ExperimentNotebook/daweilee_research/lib/python3.8/site-packages/xgboost/core.py", line 2943, in _validate_features
raise ValueError(
ValueError: training data did not have the following fields: a, b, c, d, e, f, g, h, i, j
```

Simplified Server-side log

```text
[ worker 192.168.222.236 ] : worker tcp://192.168.222.236:41425 has an empty DMatrix. [ worker 192.168.222.236 ] : 2023-11-20 16:14:37,201 - distributed.worker - WARNING - Compute Failed [ worker 192.168.222.236 ] : Key: dispatched_train-e9230b81-e512-4d69-8c95-bb15fcc7dc6b [ worker 192.168.222.236 ] : Function: dispatched_train [ worker 192.168.222.236 ] : args: ({'objective': 'reg:squarederror', 'eval_metric': ['rmse'], 'tree_method': 'hist'}, {'DMLC_NUM_WORKER': 2, 'DMLC_TRACKER_URI': '192.168.222.236', 'DMLC_TRACKER_PORT': 54469}, 139709808106176, ['Train', 'Valid'], [139709808106176, 139713269897248], {'feature_names': None, 'feature_types': None, 'feature_weights': None, 'missing': nan, 'enable_categorical': False, 'parts': [{'data': array([[ 0.11020533, 0.92716566, 0.82154456, ..., 0.18992441, [ worker 192.168.222.236 ] : 1.21373779, -0.58198202], [ worker 192.168.222.236 ] : [-1.65106638, 1.31608488, -1.47551676, ..., 0.61064058, [ worker 192.168.222.236 ] : 1.58753626, -0.05449161], [ worker 192.168.222.236 ] : [ 1.09818796, -0.04906542, 0.78345421, ..., 0.44169548, [ worker 192.168.222.236 ] : -0.3831161 , -0.20626515], [ worker 192.168.222.236 ] : ..., [ worker 192.168.222.236 ] : [-0.77153754, -0.85369089, -1.65271526, ..., -0.8760231 , [ worker 192.168.222.236 ] : 0.54686703, 0.1704825 ], [ worker 192.168.222.236 ] : [ 2.20151327, -0.28984167, -0.28146319, ..., -1.41458761, [ worker 192.168.222.236 ] : -0.55867545, -0.78704228], [ worker 192.168.222.236 ] : [ 1.33060217, -0.10747266, 0.11398802, ..., 0.15002585, [ worker 192.168.222.236 ] : 0.64218082, -0.84572091]]), 'l [ worker 192.168.222.236 ] : kwargs: {} [ worker 192.168.222.236 ] : Exception: "ValueError('Found array with 0 sample(s) (shape=(0,)) while a minimum of 1 is required.')" [ worker 192.168.222.236 ] : [ worker 192.168.222.236 ] : [16:14:37] task [xgboost.dask-tcp://192.168.222.236:41425]:tcp://192.168.222.236:41425 got new rank 1 [ worker 192.168.222.236 ] : worker tcp://192.168.222.236:41425 has an empty DMatrix. [ worker 192.168.222.236 ] : 2023-11-20 16:14:37,790 - distributed.worker - WARNING - Compute Failed [ worker 192.168.222.236 ] : Key: dispatched_train-0c3b53eb-1fdf-4603-a7af-4b0b5999b233 [ worker 192.168.222.236 ] : Function: dispatched_train [ worker 192.168.222.236 ] : args: ({'objective': 'reg:squarederror', 'eval_metric': ['rmse'], 'tree_method': 'hist'}, {'DMLC_NUM_WORKER': 2, 'DMLC_TRACKER_URI': '192.168.222.236', 'DMLC_TRACKER_PORT': 50343}, 139709765766016, ['Train', 'Valid'], [139709765766016, 139709765766880], {'feature_names': None, 'feature_types': None, 'feature_weights': None, 'missing': nan, 'enable_categorical': False, 'parts': [{'data': a b c ... h i j [ worker 192.168.222.236 ] : 0 0.110205 0.927166 0.821545 ... 0.189924 1.213738 -0.581982 [ worker 192.168.222.236 ] : 1 -1.651066 1.316085 -1.475517 ... 0.610641 1.587536 -0.054492
[ worker 192.168.222.236 ] : 2 1.098188 -0.049065 0.783454 ... 0.441695 -0.383116 -0.206265
[ worker 192.168.222.236 ] : 3 -0.108631 -0.343541 0.081264 ... -1.403026 -1.295629 0.418476
[ worker 192.168.222.236 ] : 4 -0.131076 0.790000 1.226283 ... 2.158936 1.699571 -0.617237
[ worker 192.168.222.236 ] : .. ... ... ... ... ... ... ...
[ worker 192.168.222.236 ] : 995 -0.227799 0.862158 -0.135206 ... 1.417052 0.256952 -0.069982
[ worker 192.168.222.236 ] : 996 0.803401 -0.239614 -1.651084 ... 0.690308 -0.529827 -1.85
[ worker 192.168.222.236 ] : kwargs: {}
[ worker 192.168.222.236 ] : Exception: "ValueError('training data did not have the following fields: a, b, c, d, e, f, g, h, i, j')"
[ worker 192.168.222.235 ] : 2023-11-20 16:14:37,381 - distributed.worker - WARNING - Compute Failed
[ worker 192.168.222.235 ] : Key: dispatched_train-c47ff165-96ff-4f11-b944-4b6473f5acb8
[ worker 192.168.222.235 ] : Function: dispatched_train
[ worker 192.168.222.235 ] : args: ({'objective': 'reg:squarederror', 'eval_metric': ['rmse'], 'tree_method': 'hist'}, {'DMLC_NUM_WORKER': 2, 'DMLC_TRACKER_URI': '192.168.222.236', 'DMLC_TRACKER_PORT': 54469}, 139709808106176, ['Train', 'Valid'], [139709808106176, 139713269897248], {'feature_names': None, 'feature_types': None, 'feature_weights': None, 'missing': nan, 'enable_categorical': False, 'parts': [{'data': array([[ 2.75403008, -0.02547794, 0.87412031, ..., -0.40672441,
[ worker 192.168.222.235 ] : -0.2417398 , 0.03521271],
[ worker 192.168.222.235 ] : [-1.23321979, -0.91415025, -1.67774461, ..., 0.0797747 ,
[ worker 192.168.222.235 ] : 0.43872093, -0.00441466],
[ worker 192.168.222.235 ] : [-1.66936377, 0.22684794, -0.27589295, ..., 0.67441986,
[ worker 192.168.222.235 ] : 3.12303318, 0.33610313],
[ worker 192.168.222.235 ] : ...,
[ worker 192.168.222.235 ] : [ 1.21402948, 1.16367527, -0.54400599, ..., 0.54256286,
[ worker 192.168.222.235 ] : 1.41945109, -0.34024116],
[ worker 192.168.222.235 ] : [ 0.79592589, 0.92401744, 0.88740495, ..., 0.18858176,
[ worker 192.168.222.235 ] : 1.85558868, -1.22406281],
[ worker 192.168.222.235 ] : [ 1.0047381 , -1.07957278, -0.41561333, ..., -1.17766627,
[ worker 192.168.222.235 ] : 1.98815875, -1.56373595]]), 'l
[ worker 192.168.222.235 ] : kwargs: {}
[ worker 192.168.222.235 ] : Exception: "XGBoostError('[16:14:37] /workspace/rabit/include/rabit/internal/utils.h:86: Allreduce failed')"
[ worker 192.168.222.235 ] :
[ worker 192.168.222.235 ] : [16:14:37] task [xgboost.dask-tcp://192.168.222.235:34219]:tcp://192.168.222.235:34219 got new rank 0
[ worker 192.168.222.235 ] : 2023-11-20 16:14:37,991 - distributed.worker - WARNING - Compute Failed
[ worker 192.168.222.235 ] : Key: dispatched_train-465626ff-9e7f-4ff9-94ff-a82e98a91a9d
[ worker 192.168.222.235 ] : Function: dispatched_train
[ worker 192.168.222.235 ] : args: ({'objective': 'reg:squarederror', 'eval_metric': ['rmse'], 'tree_method': 'hist'}, {'DMLC_NUM_WORKER': 2, 'DMLC_TRACKER_URI': '192.168.222.236', 'DMLC_TRACKER_PORT': 50343}, 139709765766016, ['Train', 'Valid'], [139709765766016, 139709765766880], {'feature_names': None, 'feature_types': None, 'feature_weights': None, 'missing': nan, 'enable_categorical': False, 'parts': [{'data': a b c ... h i j
[ worker 192.168.222.235 ] : 0 2.754030 -0.025478 0.874120 ... -0.406724 -0.241740 0.035213
[ worker 192.168.222.235 ] : 1 -1.233220 -0.914150 -1.677745 ... 0.079775 0.438721 -0.004415
[ worker 192.168.222.235 ] : 2 -1.669364 0.226848 -0.275893 ... 0.674420 3.123033 0.336103
[ worker 192.168.222.235 ] : 3 0.432976 -3.186649 1.148499 ... -0.981369 1.725638 1.419550
[ worker 192.168.222.235 ] : 4 -0.322792 0.295130 -0.036493 ... -1.797222 -0.509883 -0.811517
[ worker 192.168.222.235 ] : .. ... ... ... ... ... ... ...
[ worker 192.168.222.235 ] : 295 -0.435003 1.881376 0.905349 ... 1.197404 0.256697 0.405379
[ worker 192.168.222.235 ] : 296 -0.707267 -0.004189 0.686836 ... -0.599604 -1.182308 0.03
[ worker 192.168.222.235 ] : kwargs: {}
[ worker 192.168.222.235 ] : Exception: "XGBoostError('[16:14:37] /workspace/rabit/include/rabit/internal/utils.h:86: Allreduce failed')"
```

### 3. Training part code (fixed code)

```python
eval_df = dd.read_parquet('hdfs://hadoop@192.168.222.236:9000/user/lidawei/xgboost_debug/test_df.parquet')
client.rebalance([df, eval_df]) # No effect

D_Train_from_df = xgb.dask.DaskDMatrix(
client=client,
data=df[columns],
label=df['label']
)
D_Valid_from_df = xgb.dask.DaskDMatrix(
client=client,
data=eval_df[columns],
label=eval_df['label']
)

D_Train_from_array = xgb.dask.DaskDMatrix(
client=client,
data=df[columns].to_dask_array(),
label=df['label'].to_dask_array()
)
D_Valid_from_array = xgb.dask.DaskDMatrix(
client=client,
data=eval_df[columns].to_dask_array(),
label=eval_df['label'].to_dask_array()
)

# Works fine
print('=' * 10, 'Train Dask Array', '=' * 10)
print(xgb.dask.train(
client,
params={
"objective": "reg:squarederror",
"eval_metric": ["rmse"],
"tree_method": "hist",
},
dtrain=D_Train_from_array,
evals=[(D_Train_from_array, "Train"), (D_Valid_from_array, "Valid")],
num_boost_round=NUM_BOOST_ROUND,
))

print('=' * 10, 'Train Dask DataFrame', '=' * 10)
print(xgb.dask.train(
client,
params={
"objective": "reg:squarederror",
"eval_metric": ["rmse"],
"tree_method": "hist",
},
dtrain=D_Train_from_df,
evals=[(D_Train_from_df, "Train"), (D_Valid_from_df, "Valid")],
num_boost_round=NUM_BOOST_ROUND,
))

# Custom Metric
def rmse(y_pred, dtrain: xgb.DMatrix):
y_true = dtrain.get_label()
return 'custom_rmse', mean_squared_error(y_true, y_pred) ** 0.5

print('=' * 10, 'Train Dask Array with Custom Metric', '=' * 10)
try:
print(xgb.dask.train(
client,
params={
"objective": "reg:squarederror",
"eval_metric": ["rmse"],
"tree_method": "hist",
},
dtrain=D_Train_from_array,
evals=[(D_Train_from_array, "Train"), (D_Valid_from_array, "Valid")],
custom_metric=rmse,
num_boost_round=NUM_BOOST_ROUND,
))
except:
traceback.print_exc()

print('=' * 10, 'Train Dask DataFrame with Custom Metric', '=' * 10)
try:
print(xgb.dask.train(
client,
params={
"objective": "reg:squarederror",
"eval_metric": ["rmse"],
"tree_method": "hist",
},
dtrain=D_Train_from_df,
evals=[(D_Train_from_df, "Train"), (D_Valid_from_df, "Valid")],
custom_metric=rmse,
num_boost_round=NUM_BOOST_ROUND,
))
except:
traceback.print_exc()
```