Documentation: train + RayDMatrix vs XGBoostTrainer + Dataset

[andreostrovsky]

Could you please provide some clarification on the differences and/or how to choose between using xgboost_ray.train + xgboost_ray.RayDMatrix or ray.train.xgboost.XGBoostTrainer + ray.data.Dataset?

My use case is running Ray Tune on Azure Databricks, which operates on Spark. According to the Databricks docs, one creates a Ray Cluster using the Ray on Spark API, and creates a Ray Dataset from Parquet files.

Below are the questions I would like clarification on. Any help you could provide would be greatly appreciated.

Data

According to the README.md one can create a RayDMatrix from either Parquet files or a Ray Dataset:

xgboost_ray/README.md

Lines 450 to 465 in e904925

	### Data sources
	The following data sources can be used with a `RayDMatrix` object.
	| Type | Centralized loading | Distributed loading |
	|------------------------------------------------------------------|---------------------|---------------------|
	| Numpy array | Yes | No |
	| Pandas dataframe | Yes | No |
	| Single CSV | Yes | No |
	| Multi CSV | Yes | Yes |
	| Single Parquet | Yes | No |
	| Multi Parquet | Yes | Yes |
	| [Ray Dataset](https://docs.ray.io/en/latest/data/dataset.html) | Yes | Yes |
	| [Petastorm](https://github.com/uber/petastorm) | Yes | Yes |
	| [Dask dataframe](https://docs.dask.org/en/latest/dataframe.html) | Yes | Yes |
	| [Modin dataframe](https://modin.readthedocs.io/en/latest/) | Yes | Yes |

So if using xgboost_ray, should I

• Create a Ray Dataset from Parquet files, then create a RayDMatrix from that Dataset
  or
• Create the RayDMatrix directly from Parquet files

Training

Should I use Ray Tune with XGBoostTrainer or with xgboost_ray.train, running on this Ray on Spark Cluster?

I also intend to implement CV with early stopping. Since tune-sklearn is now deprecated, I understand that I'll need to implement this myself. As explained in ray-project/ray#21848 (comment), this can be done with ray.tune.stopper.TrialPlateauStopper. But according to #301 we can also use XGBoost's native xgb.callback.EarlyStopping. Which approach would you recommend? Can TrialPlateauStopper be used with xgboost_ray?

Thank you very much for any help you can offer.

[gdubs89]

I have a similar question/comment. I have been unable to find documentation or examples of how to do distributed training which uses inbuilt categorical handling. Given I want to do distributed training, the table OP posted suggests I want to be using multi-file parquet, which I am. I found an example in the docs which suggests one should be doing:

ray_params = RayDMatrix([
    "hdfs:///tmp/part1.parquet",
    "hdfs:///tmp/part2.parquet",
    "hdfs:///tmp/part3.parquet",
    "hdfs:///tmp/part4.parquet",
], label="label_col")

The problem with this however, is that if I set enable_categorical=True, it's not clear how I would specify which are the categorical columns.

I managed to get the following dask code working a while ago:

train_ddf = dd.read_parquet(s3train_filepath)
eval_ddf = dd.read_parquet(s3eval_filepath)

train_ddf[categorical_columns] = train_ddf[categorical_columns].astype('category').categorize()
category_mappings = {col: train_ddf[col].cat.categories for col in categorical_columns}
eval_ddf[categorical_columns] = eval_ddf[categorical_columns].astype('category')
for col in categorical_columns:
  eval_ddf[col] = eval_ddf[col].cat.set_categories(category_mappings[col])

dtrain = dxgb.DaskDMatrix(
    client=client,
    data=train_ddf[features],
    label=train_ddf['target'],
    enable_categorical=True)
dvalid = dxgb.DaskDMatrix(
    client=client,
    data=eval_ddf[features],
    label=eval_ddf['target'],
    enable_categorical=True)
params = {
      "objective": "binary:logistic",
      "max_depth": 5,
      'eval_metric':'logloss',
      'learning_rate':0.1,
      'tree_method':'hist',#set other hyperparameters appropriately here
  }

model = dxgb.train(
              client=client,
              params=params,
              dtrain=dtrain,
              num_boost_round=20_000,#set to a functionally infinite number to enforce early stopping
              early_stopping_rounds=10,
              evals=[(dvalid, 'eval')])

I tried something similar using the Ray categorizer, but without much luck:

train_ray = ray.data.read_parquet(s3train_filepath, columns=columns_to_keep)
eval_ray = ray.data.read_parquet(s3eval_filepath, columns=columns_to_keep)

categorizer = Categorizer(columns=categorical_columns)
train_ray = categorizer.fit_transform(train_ray)
eval_ray = categorizer.transform(eval_ray)

dtrain = RayDMatrix(train_ray.drop_columns([target_col]), label=train_ray.select_columns([target_col]), enable_categorical=True, distributed=True)
deval = RayDMatrix(eval_ray.drop_columns([target_col]), label=target_col.select([target_col]), enable_categorical=True, distributed=True)

params = {
    "objective": "reg:squarederror",  
    "eval_metric": "mae", 
    "max_depth": 10,
    "learning_rate": 0.1,
    "tree_method": "hist",  
}

results = train(
    params=params,
    dtrain=dtrain,
    evals=[(deval, "eval")],
    num_boost_round=10,  
    early_stopping_rounds=2,
    ray_params=RayParams(num_actors=256)
)

but this ultimately seems to be falling down with errors related to how the target column needs to be a string. I tried a few hacks like reading in the train features and targets separately but that also didn't work. I am also trying to run this on databricks.

So my question is, is the workflow of "read in data, define categorical columns, create dmatrices with enable_categorical=True and then fit, keeping the whole pipeline distributed" something currently supported and if so, what's the canonical way of doing it?