# Understanding distributed data loading and training xgboost ray

**URL:** <https://discuss.ray.io/t/understanding-distributed-data-loading-and-training-xgboost-ray/11271>\
**Category:** Ray Data\
**Created:** [July 5, 2023, 10:54am UTC](https://discuss.ray.io/t/understanding-distributed-data-loading-and-training-xgboost-ray/11271 "2023-07-05T10:54:35Z")\
**Posts on this page:** 11\
**Page:** 1

<div class="post-metadata">

**Author:** ![Y\_C](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ray.io/y_c/32/4142_2.png) [@Y\_C](https://discuss.ray.io/u/Y_C)\
**Post date:** [July 5, 2023, 10:54am UTC](https://discuss.ray.io/t/understanding-distributed-data-loading-and-training-xgboost-ray/11271/1 "2023-07-05T10:54:35Z")

</div>

**How severe does this issue affect your experience of using Ray?**

- High: It blocks me to complete my task.

I’m using xgboost\_ray package to train an xgboost model.  
The data is loaded from ray.Dataset to RayDMatrix. I’m giving 2 actors and 1 cpu per actor as RayParams.

```auto
def load_data(path: str) -> RayDMatrix:
    ds = ray.data.read_parquet(path)
    label_col = ds.schema().names[-1] # User needs to ensure that the last column is the label column
    return RayDMatrix(ds, label=label_col)

```

The path is a directory in google cloud storage. The training process:

```auto
@remote(num_cpu=1, max_task_retries=3, max_restarts=3)
def MyTrainer:
    def training_pipeline():
        train_dtrain = load_data("gs://my_bucket/my_dir_with_parquet_files")
        xgb_reg_model = xgboost_ray.train(param, train_dtrain, RayParams(num_actors=2, cpus_per_actor=1), early_stopping_rounds=int(param['early_stopping_rounds']), num_boost_round=int(param['n_estimators']), evals=[(train_dtrain, 'train')],
                                           evals_result=result, verbose_eval=True, xgb_model=model)
my_trainer_instance = MyTrainer()
ray_ref = my_training_instance.training_pipeline.remote()
ray.get(ray_ref)

```

Got the error:

```auto
INFO: __main__ :ray job finished with state: FAILED
ERROR: __main__ :
 error trace back: Job failed due to an application error, last available logs (truncated to 20,000 chars):
  File "/home/ray/anaconda3/lib/python3.10/site-packages/xgboost_ray/main.py", line 1598, in train
    bst, train_evals_result, train_additional_results = _train(
  File "/home/ray/anaconda3/lib/python3.10/site-packages/xgboost_ray/main.py", line 1134, in _train
    dtrain.assert_enough_shards_for_actors(num_actors=ray_params.num_actors)
  File "/home/ray/anaconda3/lib/python3.10/site-packages/xgboost_ray/matrix.py", line 884, in assert_enough_shards_for_actors
    self.loader.assert_enough_shards_for_actors(num_actors=num_actors)
  File "/home/ray/anaconda3/lib/python3.10/site-packages/xgboost_ray/matrix.py", line 569, in assert_enough_shards_for_actors
    raise RuntimeError(
RuntimeError: Trying to shard data for 2 actors, but the maximum number of shards is 1. If you want to shard the dataset by rows, conside
r centralized loading by passing `distributed=False` to the `RayDMatrix`. Otherwise consider using fewer actors or re-partitioning your d
ata.

```

Questions:

1. I understood my error since there is only one parquet file in my directory. : [https://github.com/ray-project/xgboost\_ray:](https://github.com/ray-project/xgboost_ray:)  
`Centralized loading is used when you pass centralized in-memory dataframes, such as Pandas dataframes ... such as a single CSV or Parquet file.`  
What if I first read the file into a ray Dataset and then load it into RayDMatrix? Does it still use centralised loading?

2. It seems that it’s required to give a list of file names to enable distributed loading, even if a directory contains multiple parquet files is given, it’s still using centralised data loading.

3. Should I directly read parquet files to RayDMatrix instead of using RayDataset as intermediate? However since I don’t know the label\_col before hand, I used ray dataset to get the column names.

4. Does centralized loading mean that it will require more memory or is it slower? And if using centralized loading, how many shards will the head node create, is it the number of actors given in the train func?

5. What is the relationship between the cpu\_per\_actor and the resource requested by the ray actor (MyTrainer) which contains the training logic, where I have the `@remote(num_cpu=1)`?  
If I use `RayParams(num_actors=2, cpus_per_actor=1)` in the train function and `@remote(num_cpu=1)` on the MyTrainer, it seems that in total 3 actors are created. What would be the logical resource required to schedule this training job? 3 cpus?

 ![Screenshot 2023-07-07 at 18.02.01](https://us1.discourse-cdn.com/flex020/uploads/ray/original/2X/6/6da2f6ce84c1abb5ca46becd3e42b1943933565f.png)

1. Another thing I noticed was that scaling up was triggered at almost the end of the training job, it scaled up to 5 worker nodes from 1 worker node, although there is enough cpus on my worker node. And all the new workers got killed very quickly since the training job is done. It seemed that it was retrying something? Any suggestions to avoid the unneeded scale up?

`103e[2me[36m(_RemoteRayXGBoostActor pid=1050, ip=10.52.5.2)e[0m 2023-07-07 08:31:31,139	INFO main.py:1254 -- Training in progress (31 seconds since last restart).`  
And I noticed there is one task that failed in each of the training actors.

 ![Screenshot 2023-07-07 at 18.12.50](https://us1.discourse-cdn.com/flex020/uploads/ray/original/2X/a/a85ecf59269361528fdd03dcde4173638db6a34e.png)

---

<div class="post-metadata">

**Author:** ![chengsu](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ray.io/chengsu/32/2922_2.png) [@chengsu](https://discuss.ray.io/u/chengsu)\
**Post date:** [July 7, 2023, 10:54pm UTC](https://discuss.ray.io/t/understanding-distributed-data-loading-and-training-xgboost-ray/11271/2 "2023-07-07T22:54:59Z")

</div>

Hi @Y_C, re error of

```auto
RuntimeError: Trying to shard data for 2 actors, but the maximum number of shards is 1. If you want to shard the dataset by rows, conside
r centralized loading by passing `distributed=False` to the `RayDMatrix`. Otherwise consider using fewer actors or re-partitioning your d
ata.

```

As you said, yes, this is because you only has 1 Parquet file in directory. For this example that you have 2 actors / training workers, you can call `Dataset.repartition(2)` before creating the `RayDMatrix`, so the Ray Dataset you created would have two blocks / shards, and each actor can train on.

```auto
def load_data(path: str) -> RayDMatrix:
    ds = ray.data.read_parquet(path)
    ds = ds.repartition(2)
    label_col = ds.schema().names[-1] # User needs to ensure that the last column is the label column
    return RayDMatrix(ds, label=label_col)

```

Note: you don’t need to do the `Dataset.repartition()` if you have multiple Parquet files in your directory.

---

<div class="post-metadata">

**Author:** ![chengsu](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ray.io/chengsu/32/2922_2.png) [@chengsu](https://discuss.ray.io/u/chengsu)\
**Post date:** [July 7, 2023, 11:01pm UTC](https://discuss.ray.io/t/understanding-distributed-data-loading-and-training-xgboost-ray/11271/3 "2023-07-07T23:01:19Z")

</div>

> [@Y\_C](#):
>
> What is the relationship between the cpu\_per\_actor and the resource requested by the ray actor (MyTrainer) which contains the training logic, where I have the `@remote(num_cpu=1)`?  
> If I use `RayParams(num_actors=2, cpus_per_actor=1)` in the train function and `@remote(num_cpu=1)` on the MyTrainer, it seems that in total 3 actors are created. What would be the logical resource required to schedule this training job? 3 cpus?

Yes it would be 3 actors and each one requires 1 CPU logical resource. One actor for the `MyTrainer ` class, and two actors for training workers, each one of them is `_RemoteRayXGBoostActor`. Is there a reason why you need to create the `MyTrainer` class? It seems to me you can call `xgboost_ray.train()` directly, without wrapping it inside `MyTrainer.training_pipeline()`.

Note the `_StatsActor`, `_QueueActor` and `_EventActor` are internal actors created during Ray execution, not related to your application code.

---

<div class="post-metadata">

**Author:** ![Y\_C](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ray.io/y_c/32/4142_2.png) [@Y\_C](https://discuss.ray.io/u/Y_C)\
**Post date:** [July 13, 2023, 11:06am UTC](https://discuss.ray.io/t/understanding-distributed-data-loading-and-training-xgboost-ray/11271/4 "2023-07-13T11:06:06Z")

</div>

Thanks so much for replying @chengsu . The reason is that we are creating a training platform that can work with different framework (sklearn, xgboost etc) and needs the user to write the training script in a certain structure. We might look into removing the unneeded extra actor. Should I just remove the `@remote(num_cpu=1)` on the MyTrainer actor? If I understand correctly this is only for the auto scaler to determine whether the ray cluster needs to add more worker right?

Can you please also have a look at my question 6? I suspect that is related to the retried tasks, but I’m not sure how to verify that, do you have suggestions? It seems the failed tasks don’t really matter, it’s ray termination? The job still succeeded.

---

<div class="post-metadata">

**Author:** ![chengsu](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ray.io/chengsu/32/2922_2.png) [@chengsu](https://discuss.ray.io/u/chengsu)\
**Post date:** [July 14, 2023, 1:22am UTC](https://discuss.ray.io/t/understanding-distributed-data-loading-and-training-xgboost-ray/11271/5 "2023-07-14T01:22:49Z")

</div>

> [@Y\_C](#):
>
> Should I just remove the `@remote(num_cpu=1)` on the MyTrainer actor? If I understand correctly this is only for the auto scaler to determine whether the ray cluster needs to add more worker right?

Yes, just remove the `@remote(num_cpu=1)` on the MyTrainer actor should be good.

---

<div class="post-metadata">

**Author:** ![chengsu](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ray.io/chengsu/32/2922_2.png) [@chengsu](https://discuss.ray.io/u/chengsu)\
**Post date:** [July 14, 2023, 1:26am UTC](https://discuss.ray.io/t/understanding-distributed-data-loading-and-training-xgboost-ray/11271/6 "2023-07-14T01:26:07Z")

</div>

> [@Y\_C](#):
>
> Can you please also have a look at my question 6? I suspect that is related to the retried tasks, but I’m not sure how to verify that, do you have suggestions? It seems the failed tasks don’t really matter, it’s ray termination? The job still succeeded.

I feel this is coming from task retry. Here is our doc to understand dashboard - [Ray Dashboard — Ray 2.8.0](https://docs.ray.io/en/latest/ray-observability/getting-started.html) - “Task and Actor errors”.

---

<div class="post-metadata">

**Author:** ![kai](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ray.io/kai/32/3380_2.png) [@kai](https://discuss.ray.io/u/kai)\
**Post date:** [July 17, 2023, 4:52pm UTC](https://discuss.ray.io/t/understanding-distributed-data-loading-and-training-xgboost-ray/11271/7 "2023-07-17T16:52:46Z")

</div>

For the original issue, this was a bug in XGBoost-Ray - with a Ray dataset, we can repartition a dataset automatically if needed. This was fixed in xgboost\_ray==0.17.0 and the issue shouldn’t come up anymore.

---

<div class="post-metadata">

**Author:** ![Y\_C](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ray.io/y_c/32/4142_2.png) [@Y\_C](https://discuss.ray.io/u/Y_C)\
**Post date:** [July 17, 2023, 4:56pm UTC](https://discuss.ray.io/t/understanding-distributed-data-loading-and-training-xgboost-ray/11271/8 "2023-07-17T16:56:45Z")

</div>

> [@Y\_C](#):
>
> y in google cloud storage. The training process:

Thanks for the reply. Do you mean that now even if reading from a single file it would be loaded in a distributed way? Or only if I first load it into a ray dataset?  
Can you also please elaborate the difference between centralised loading and distributed loading? It seems that if the centralized loader is chosen, it would still require the machine to be able to hold the entire dataset in memory, cause underneath it loads the data into pandas dataframe. Is it correct?

---

<div class="post-metadata">

**Author:** ![kai](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ray.io/kai/32/3380_2.png) [@kai](https://discuss.ray.io/u/kai)\
**Post date:** [July 19, 2023, 7:34am UTC](https://discuss.ray.io/t/understanding-distributed-data-loading-and-training-xgboost-ray/11271/9 "2023-07-19T07:34:20Z")

</div>

Yes that’s right. Centralized loading is useful if the data is small enough and you have it available on one machine. It will then distribute the data from a central location. The machine needs to be able to hold the full dataset in memory.

In most cases distributed data loading should be preferred if possible.

Regarding your last question, yes, the fix affects Ray Datasets, so if you use them to load the data, it will be distributed automatically. Note that if it’s only one parquet file this will be similar to the centralized data loading in XGBoost-Ray.

---

<div class="post-metadata">

**Author:** ![Y\_C](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ray.io/y_c/32/4142_2.png) [@Y\_C](https://discuss.ray.io/u/Y_C)\
**Post date:** [July 19, 2023, 7:58am UTC](https://discuss.ray.io/t/understanding-distributed-data-loading-and-training-xgboost-ray/11271/10 "2023-07-19T07:58:20Z")

</div>

@kai thanks! So the fix is for the case where a directory containing more than one files right? So the partition depends on if the data is already split into multiple files. Can you please share a link to the github issue of this fix? Thanks!

---

<div class="post-metadata">

**Author:** ![kai](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.ray.io/kai/32/3380_2.png) [@kai](https://discuss.ray.io/u/kai)\
**Post date:** [July 19, 2023, 8:54am UTC](https://discuss.ray.io/t/understanding-distributed-data-loading-and-training-xgboost-ray/11271/11 "2023-07-19T08:54:46Z")

</div>

Sure, the pull request is here: [Repartition Ray dataset if number of shards is too small by krfricke · Pull Request #283 · ray-project/xgboost\_ray · GitHub](https://github.com/ray-project/xgboost_ray/pull/283)

Yes, if you have more than one file this will be the number of partitions in XGBoost-Ray or the number of blocks in a Ray Dataset.