[FEA] get dask_cudf.Series from xgb.dask.predict()

[rnyak]

Is your feature request related to a problem? Please describe.
I would like to obtain prediction result from prediction = xgb.dask.predict(client, output, dtrain) as dask_cudf.Series instead of dask.array.core.Array .

I am using, rapids 0.13 nightly in a conda env, and Dask 2.12.0. Here is a min rep example code:

cdf = cudf.DataFrame()
cdf['day'] = [15, 10, 20, 20,  21, 25, 28, 29]
cdf['hour'] = [19, 20, 20, 21, 18, 12, 15, 13]
cdf['passenger_count'] = [1, 1, 2, 2, 3, 3, 4, 2]
cdf['fare_amount'] = [5.0, 3.5, 12.5, 4.5, 9.0, 5.0, 3.5, 7.5]
ddf=dask_cudf.from_cudf(cdf, npartitions=2)

%%time
X_train = ddf.query('day < 25').persist()

# create a Y_train ddf with just the target variable
Y_train = X_train[['fare_amount']].persist()
# drop the target variable from the training ddf
X_train = X_train[X_train.columns.difference(['fare_amount'])]

# this wont return until all data is in GPU memory
done = wait([X_train, Y_train])

import xgboost as xgb
dtrain = xgb.dask.DaskDMatrix(client, X_train, Y_train)

%%time
trained_model = xgb.dask.train(client,
                        {
                         'learning_rate': 0.3,
                          'max_depth': 8,
                          'objective': 'reg:squarederror',
                          'subsample': 0.6,
                          'gamma': 1,
                          'silent': True,
                          'verbose_eval': True,
                          'tree_method':'gpu_hist',
                          'n_gpus': 1
                        },
                        dtrain,
                        num_boost_round=100, evals=[(dtrain, 'train')])
def drop_empty_partitions(df):
    lengths = df.map_partitions(len).compute()
    nonempty = [length > 0 for length in lengths]
    return df.partitions[nonempty]

X_test = ddf.query('day >= 25').persist()
X_test = drop_empty_partitions(X_test)

# Create Y_test with just the fare amount
Y_test = X_test[['fare_amount']]

# Drop the fare amount from X_test
X_test = X_test[X_test.columns.difference(['fare_amount'])]

dtest = xgb.dask.DaskDMatrix(client, X_test, Y_test)
prediction = xgb.dask.predict(client, trained_model['booster'], dtest)
type(prediction)
dask.array.core.Array

Describe the solution you'd like

prediction = xgb.dask.predict(client, output, dtrain)
type(prediction)
dask_cudf.core.Series

# I'd like to be able to calculate RMSE error as follows:
Y_test['squared_error'] = (prediction- Y_test['fare_amount'])**2
math.sqrt(Y_test.squared_error.mean().compute())

[rnyak]

Moved to dmlcx/xgboost