diff --git a/cpp/include/cuml/ensemble/randomforest_mg_utils.hpp b/cpp/include/cuml/ensemble/randomforest_mg_utils.hpp new file mode 100644 index 0000000000..302aa43d71 --- /dev/null +++ b/cpp/include/cuml/ensemble/randomforest_mg_utils.hpp @@ -0,0 +1,26 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#pragma once + +#include +#include +#include +#include + +namespace ML::detail { + +inline void cuml_rf_allreduce_validation_status(const raft::handle_t& handle, + const int* local_status, + int* global_status) +{ + auto const& comm = raft::resource::get_comms(handle); + cudaStream_t stream = raft::resource::get_cuda_stream(handle); + comm.allreduce(local_status, global_status, 1, raft::comms::op_t::MAX, stream); + RAFT_EXPECTS(comm.sync_stream(stream) == raft::comms::status_t::SUCCESS, + "Input validation status all-reduce failed"); +} + +} // namespace ML::detail diff --git a/cpp/src/randomforest/randomforest.cuh b/cpp/src/randomforest/randomforest.cuh index b44c1d3c80..4bfab28807 100644 --- a/cpp/src/randomforest/randomforest.cuh +++ b/cpp/src/randomforest/randomforest.cuh @@ -330,10 +330,12 @@ class RandomForest { // Distributed tree builders issue collectives independently, so train them serially until // the forest-level scheduler can impose a global collective order across concurrent trees. if (distributed) { n_streams = 1; } - ASSERT(static_cast(n_streams) <= handle.get_stream_pool_size(), - "effective RF n_streams (=%d) should be <= raft::handle_t.n_streams (=%lu)", - n_streams, - handle.get_stream_pool_size()); + auto stream_pool_size = handle.get_stream_pool_size(); + if (static_cast(n_streams) > stream_pool_size) { + CUML_LOG_WARN("Resizing n_streams to fit the available stream pool size (%lu)", + stream_pool_size); + n_streams = ML::narrow_cast(stream_pool_size); + } auto quantile_result = DT::computeQuantiles(handle, input, diff --git a/python/cuml/cuml/dask/ensemble/base.py b/python/cuml/cuml/dask/ensemble/base.py index 65b01eb67a..eac87d180a 100644 --- a/python/cuml/cuml/dask/ensemble/base.py +++ b/python/cuml/cuml/dask/ensemble/base.py @@ -1,19 +1,13 @@ # SPDX-FileCopyrightText: Copyright (c) 2021-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 # - -import math import warnings -from collections.abc import Iterable -import cupy as cp -import dask -import numpy as np -import treelite -from dask.distributed import Future +from dask.distributed import get_worker +from raft_dask.common.comms import Comms, get_raft_comm_state -from cuml import using_output_type from cuml.dask._compat import DASK_2025_4_0 +from cuml.dask.common.base import mnmg_import from cuml.dask.common.input_utils import DistributedDataHandler, concatenate from cuml.dask.common.utils import get_client, wait_and_raise_from_futures @@ -46,180 +40,95 @@ def _create_model( ) self.workers = workers self._set_internal_model(None) - self.active_workers = list() - self.ignore_empty_partitions = ignore_empty_partitions self.n_estimators = n_estimators - self.n_estimators_per_worker = self._estimators_per_worker( - n_estimators - ) - if base_seed is None: - base_seed = 0 - seeds = [base_seed] - for i in range(1, len(self.n_estimators_per_worker)): - sd = self.n_estimators_per_worker[i - 1] + seeds[i - 1] - seeds.append(sd) + if "n_streams" in kwargs: + warnings.warn( + ( + "n_streams has no effect on distributed training and " + "will be removed in release 26.12." + ), + FutureWarning, + stacklevel=2, + ) + if ignore_empty_partitions is not None: + warnings.warn( + ( + "ignore_empty_partitions parameter is no longer valid " + "and will be removed in release 26.12." + ), + FutureWarning, + stacklevel=2, + ) self.rfs = { worker: self.client.submit( model_func, - n_estimators=self.n_estimators_per_worker[n], - random_state=seeds[n], + n_estimators=self.n_estimators, + random_state=base_seed, **kwargs, pure=False, workers=[worker], ) - for n, worker in enumerate(self.workers) + for worker in self.workers } wait_and_raise_from_futures(list(self.rfs.values())) - def _estimators_per_worker(self, n_estimators): - n_workers = len(self.workers) - if n_estimators < n_workers: - raise ValueError( - "n_estimators cannot be lower than number of dask workers." - ) - - n_est_per_worker = math.floor(n_estimators / n_workers) - n_estimators_per_worker = [n_est_per_worker for i in range(n_workers)] - remaining_est = n_estimators - (n_est_per_worker * n_workers) - for i in range(remaining_est): - n_estimators_per_worker[i] = n_estimators_per_worker[i] + 1 - return n_estimators_per_worker - - def _fit(self, model, dataset, broadcast_data): + def _fit(self, model, dataset, classes=None): data = DistributedDataHandler.create(dataset, client=self.client) - self.active_workers = data.workers self.datatype = data.datatype - labels = self.client.persist(dataset[1]) - if self.datatype == "cudf": - self.num_classes = len(labels.unique()) - else: - self.num_classes = len(dask.array.unique(labels).compute()) - - combined_data = ( - list(map(lambda x: x[1], data.gpu_futures)) - if broadcast_data - else None + unknown_workers = set(data.workers).difference(model) + if unknown_workers: + raise ValueError( + "Training data was placed on workers that were not selected " + f"for this estimator: {sorted(unknown_workers)}" + ) + if not data.worker_to_parts: + raise ValueError("No mapping found between workers and partitions") + + total_rows = sum(total for _, total in data._worker_sizes.values()) + comms = Comms( + comms_p2p=False, + client=self.client, + streams_per_handle=1, ) - - futures = list() - for idx, (worker, worker_data) in enumerate( - data.worker_to_parts.items() - ): - futures.append( - self.client.submit( + futures = [] + try: + comms.init(workers=data.workers) + for worker, worker_data in data.worker_to_parts.items(): + future = self.client.submit( _func_fit, + comms.sessionId, model[worker], - combined_data if broadcast_data else worker_data, + worker_data, + total_rows, + classes, workers=[worker], pure=False, ) - ) + futures.append(future) + self.rfs[worker] = future - self.n_active_estimators_per_worker = [] - for worker in data.worker_to_parts.keys(): - n = self.workers.index(worker) - n_est = self.n_estimators_per_worker[n] - self.n_active_estimators_per_worker.append(n_est) + wait_and_raise_from_futures(futures) + finally: + comms.destroy() - if len(self.workers) > len(self.active_workers): - if self.ignore_empty_partitions: - curent_estimators = ( - self.n_estimators - / len(self.workers) - * len(self.active_workers) - ) - warn_text = ( - f"Data was not split among all workers " - f"using only {self.active_workers} workers to fit." - f"This will only train {curent_estimators}" - f" estimators instead of the requested " - f"{self.n_estimators}" - ) - warnings.warn(warn_text) - else: - raise ValueError( - "Data was not split among all workers. " - "Re-run the code or " - "use ignore_empty_partitions=True" - " while creating model" - ) - wait_and_raise_from_futures(futures) + # Every distributed rank owns the same complete forest. Keep one + # worker future as the canonical model for inference and serialization. + self._set_internal_model(futures[0]) return self - def _concat_treelite_models(self): - """ - Convert the cuML Random Forest model present in different workers to - the treelite format and then concatenate the different treelite models - to create a single model. The concatenated model is then converted to - bytes format. - """ - model_serialized_futures = list() - for w in self.active_workers: - model_serialized_futures.append( - dask.delayed(_serialize_treelite_bytes)(self.rfs[w]) - ) - mod_bytes = self.client.compute(model_serialized_futures, sync=True) - last_worker = w - model = self.rfs[last_worker].result() - tl_model_objs = [ - treelite.Model.deserialize_bytes(indiv_worker_model_bytes) - for indiv_worker_model_bytes in mod_bytes - ] - concatenated_model = treelite.Model.concatenate(tl_model_objs) - model._treelite_model_bytes = concatenated_model.serialize_bytes() - model._fil_model = None - return model - - def _partial_inference(self, X, op_type, delayed, **kwargs): + def _predict_using_nvforest(self, X, delayed, **kwargs): data = DistributedDataHandler.create(X, client=self.client) - combined_data = list(map(lambda x: x[1], data.gpu_futures)) - - if op_type == "classification": - func = _func_predict_proba_partial - shape = (X.shape[0], 1, self.num_classes) - else: - shape = (X.shape[0], 1) - func = _func_predict_partial - - meta = cp.zeros((0,) * len(shape), dtype=cp.float32) - - partial_infs = list() - for worker in self.active_workers: - partial_infs.append( - self.client.submit( - func, - self.rfs[worker], - combined_data, - **kwargs, - workers=[worker], - pure=False, - ) - ) - - objs = [ - dask.array.from_delayed(partial_inf, shape=shape, meta=meta) - for partial_inf in partial_infs - ] - result = dask.array.concatenate(objs, axis=1) - return result - - def _predict_using_fil(self, X, delayed, **kwargs): - if self._get_internal_model() is None: - self._set_internal_model(self._concat_treelite_models()) - data = DistributedDataHandler.create(X, client=self.client) - if self._get_internal_model() is None: - self._set_internal_model(self._concat_treelite_models()) return self._predict( X, delayed=delayed, output_collection_type=data.datatype, **kwargs ) def _get_params(self, deep): model_params = list() - for idx, worker in enumerate(self.workers): + for worker in self.workers: model_params.append( self.client.submit( _func_get_params, self.rfs[worker], deep, workers=[worker] @@ -229,8 +138,17 @@ def _get_params(self, deep): return params_of_each_model def _set_params(self, **params): + if "n_streams" in params: + warnings.warn( + ( + "n_streams has no effect on distributed training and " + "will be removed in release 26.12." + ), + FutureWarning, + stacklevel=2, + ) model_params = list() - for idx, worker in enumerate(self.workers): + for worker in self.workers: model_params.append( self.client.submit( _func_set_params, @@ -242,96 +160,34 @@ def _set_params(self, **params): wait_and_raise_from_futures(model_params) return self - def get_combined_model(self): - """ - Return single-GPU model for serialization. - - Returns - ------- - - model : Trained single-GPU model or None if the model has not - yet been trained. - """ - - # set internal model if it hasn't been accessed before - if self._get_internal_model() is None: - self._set_internal_model(self._concat_treelite_models()) - - internal_model = self._check_internal_model(self._get_internal_model()) - - if isinstance(self.internal_model, Iterable): - # This function needs to return a single instance of cuml.Base, - # even if the class is just a composite. - raise ValueError( - "Expected a single instance of cuml.Base " - "but got %s instead." % type(self.internal_model) - ) - - elif isinstance(self.internal_model, Future): - internal_model = self.internal_model.result() - - return internal_model - - def _get_workers_weights(self) -> cp.ndarray: - workers_weights = np.array(self.n_active_estimators_per_worker) - workers_weights = workers_weights[workers_weights != 0] - workers_weights = workers_weights / workers_weights.sum() - workers_weights = cp.array(workers_weights) - return workers_weights - - def apply_reduction(self, reduce, partial_infs, datatype, delayed): - """ - Reduces the partial inferences to obtain the final result. The workers - didn't have the same number of trees to form their predictions. To - correct for this worker's predictions are weighted differently during - reduction. - """ - workers_weights = self._get_workers_weights() - unique_classes = ( - None - if not hasattr(self, "unique_classes") - else self.unique_classes - ) - delayed_local_array = dask.delayed(reduce)( - partial_infs, workers_weights, unique_classes - ) - delayed_res = dask.array.from_delayed( - delayed_local_array, shape=(np.nan, np.nan), dtype=np.float32 - ) - if delayed: - return delayed_res - else: - return delayed_res.persist() - -def _func_fit(model, input_data): +@mnmg_import +def _func_fit(session_id, model, input_data, total_rows, classes): + handle = get_raft_comm_state(session_id, get_worker())["handle"] X = concatenate([item[0] for item in input_data]) y = concatenate([item[1] for item in input_data]) - return model.fit(X, y) - - -def _func_predict_partial(model, input_data, **kwargs): - """ - Whole dataset inference with part of the model (trees at disposal locally). - Transfer dataset instead of model. Interesting when model is larger - than dataset. - """ - X = concatenate(input_data) - with using_output_type("cupy"): - prediction = model.predict(X, **kwargs) - return cp.expand_dims(prediction, axis=1) - - -def _func_predict_proba_partial(model, input_data, **kwargs): - """ - Whole dataset inference with part of the model (trees at disposal locally). - Transfer dataset instead of model. Interesting when model is larger - than dataset. - """ - X = concatenate(input_data) - with using_output_type("cupy"): - prediction = model.predict_proba(X, **kwargs) - return cp.expand_dims(prediction, axis=1) + model._raft_handle = handle + model._distributed_n_rows = total_rows + if classes is not None: + model._distributed_classes = classes + try: + validation_error = None + try: + model._prepare_fit_inputs(X, y) + except Exception as error: + validation_error = error + + if model._allreduce_validation_status(validation_error is not None): + if validation_error is not None: + raise validation_error + raise RuntimeError("Input validation failed on another worker") + + return model.fit(X, y) + finally: + del model._raft_handle + del model._distributed_n_rows + if classes is not None: + del model._distributed_classes def _func_get_params(model, deep): @@ -340,7 +196,3 @@ def _func_get_params(model, deep): def _func_set_params(model, **params): return model.set_params(**params) - - -def _serialize_treelite_bytes(model): - return model._treelite_model_bytes diff --git a/python/cuml/cuml/dask/ensemble/randomforestclassifier.py b/python/cuml/cuml/dask/ensemble/randomforestclassifier.py index 97b9aca4d7..d28ed39ae7 100755 --- a/python/cuml/cuml/dask/ensemble/randomforestclassifier.py +++ b/python/cuml/cuml/dask/ensemble/randomforestclassifier.py @@ -2,6 +2,8 @@ # SPDX-FileCopyrightText: Copyright (c) 2019-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 # +import warnings + import cupy as cp import dask.array @@ -26,30 +28,13 @@ class RandomForestClassifier( classifiers in an ensemble. This uses Dask to partition data over multiple GPUs (possibly on different nodes). - This implementation makes the following assumptions: - * The set of Dask workers used between instantiation, fit, \ - and predict are all consistent - * Training data comes in the form of cuDF dataframes or Dask Arrays \ - distributed so that each worker has at least one partition. - - The distributed algorithm uses an *embarrassingly-parallel* - approach. For a forest with `N` trees being built on `w` workers, each - worker simply builds `N/w` trees on the data it has available - locally. In many cases, partitioning the data so that each worker - builds trees on a subset of the total dataset works well, but - it generally requires the data to be well-shuffled in advance. - Alternatively, callers can replicate all of the data across - workers so that ``rf.fit`` receives `w` partitions, each containing the - same data. This would produce results approximately identical to - single-GPU fitting. - - Please check the single-GPU implementation of Random Forest - classifier for more information about the underlying algorithm. + During fitting, all workers that hold training rows collectively build the + same forest from the complete distributed dataset. Parameters ---------- n_estimators : int (default = 100) - total number of trees in the forest (not per-worker) + total number of trees in the forest split_criterion : int or string (default = ``0`` (``'gini'``)) The criterion used to split nodes.\n * ``0`` or ``'gini'`` for gini impurity @@ -106,20 +91,17 @@ class RandomForestClassifier( * If type ``float``, then ``min_samples_split`` represents a fraction and ``ceil(min_samples_split * n_rows)`` is the minimum number of samples for each split. - - n_streams : int (default = 4 ) - Number of parallel streams used for forest building + n_streams : int + Deprecated. Distributed training currently builds trees serially to + preserve collective order. workers : optional, list of strings Dask addresses of workers to use for computation. If None, all available Dask workers will be used. random_state : int (default = None) Seed for the random number generator. Unseeded by default. - - ignore_empty_partitions: Boolean (default = False) - Specify behavior when a worker does not hold any data - while splitting. When True, it returns the results from workers - with data (the number of trained estimators will be less than - n_estimators) When False, throws a RuntimeError. + ignore_empty_partitions: optional, boolean + Deprecated. This parameter no longer has any effect and + will be removed in release 26.12. Examples -------- @@ -135,7 +117,7 @@ def __init__( verbose=False, n_estimators=100, random_state=None, - ignore_empty_partitions=False, + ignore_empty_partitions=None, **kwargs, ): super().__init__(client=client, verbose=verbose, **kwargs) @@ -155,12 +137,11 @@ def _construct_rf(n_estimators, random_state, **kwargs): n_estimators=n_estimators, random_state=random_state, **kwargs ) - def fit(self, X, y, broadcast_data=False): + def fit(self, X, y, broadcast_data=None): """ Fit the input data with a Random Forest classifier - IMPORTANT: X is expected to be partitioned with at least one partition - on each Dask worker being used by the forest (self.workers). + Only workers holding one or more training rows participate in fitting. If a worker has multiple data partitions, they will be concatenated before fitting, which will lead to additional memory usage. To minimize @@ -196,27 +177,30 @@ def fit(self, X, y, broadcast_data=False): y : Dask cuDF dataframe or CuPy backed Dask Array (n_rows, 1) Labels of training examples. **y must be partitioned the same way as X** - broadcast_data : bool, optional (default = False) - When set to True, the whole dataset is broadcasted - to train the workers, otherwise each worker - is trained on its partition + broadcast_data : bool, optional + Deprecated. This parameter no longer has effect and will + be removed in release 26.12. """ - # Handle both Dask Arrays and Dask Series/DataFrames + if broadcast_data is not None: + warnings.warn( + ( + "broadcast_data parameter is no longer valid " + "and will be removed in release 26.12." + ), + FutureWarning, + stacklevel=2, + ) if isinstance(y, dask.array.Array): - # For Dask Arrays, use dask.array.unique unique_vals = dask.array.unique(y).compute() - self.unique_classes = cp.sort(cp.asarray(unique_vals)) else: - # For Dask Series/DataFrames, use .unique() method - self.unique_classes = cp.asarray( - y.unique().compute().sort_values(ignore_index=True) - ) - self.num_classes = len(self.unique_classes) + unique_vals = y.unique().compute().sort_values(ignore_index=True) + classes = cp.asnumpy(cp.sort(cp.asarray(unique_vals))) + self.classes_ = classes self._set_internal_model(None) self._fit( model=self.rfs, dataset=(X, y), - broadcast_data=broadcast_data, + classes=classes, ) return self @@ -228,7 +212,7 @@ def predict( default_chunk_size=None, align_bytes=None, delayed=True, - broadcast_data=False, + broadcast_data=None, ): """ Predicts the labels for X. @@ -241,7 +225,7 @@ def predict( threshold : float (default = 0.5) Threshold used for classification. layout : string (default = 'depth_first') - Specifies the in-memory layout of nodes in FIL forests. Options: + Specifies the in-memory layout of nodes in nvForest models. Options: 'depth_first', 'layered', 'breadth_first'. default_chunk_size : int, optional (default = None) Determines how batches are further subdivided for parallel processing. @@ -255,28 +239,25 @@ def predict( delayed : bool (default = True) Whether to do a lazy prediction (and return Delayed objects) or an eagerly executed one. - broadcast_data : bool (default = False) - If False, the trees are merged in a single model before the workers - perform inference on their share of the prediction workload. - When True, trees aren't merged. Instead each worker infers on the - whole prediction workload using its available trees. The results are - reduced on the client. May be advantageous when the model is larger - than the data used for inference. + broadcast_data : bool, optional + Deprecated. This parameter no longer has effect and will + be removed in release 26.12. Returns ------- y : Dask cuDF dataframe or CuPy backed Dask Array (n_rows, 1) The predicted class labels. """ - if broadcast_data: - return self.partial_inference( - X, - layout=layout, - default_chunk_size=default_chunk_size, - align_bytes=align_bytes, - delayed=delayed, + if broadcast_data is not None: + warnings.warn( + ( + "broadcast_data parameter is no longer valid " + "and will be removed in release 26.12." + ), + FutureWarning, + stacklevel=2, ) - return self._predict_using_fil( + return self._predict_using_nvforest( X, threshold=threshold, layout=layout, @@ -285,26 +266,6 @@ def predict( delayed=delayed, ) - def partial_inference(self, X, delayed, **kwargs): - partial_infs = self._partial_inference( - X=X, op_type="classification", delayed=delayed, **kwargs - ) - worker_weights = self._get_workers_weights() - merged_votes = dask.array.average( - partial_infs, axis=1, weights=worker_weights - ) - pred_class_indices = merged_votes.argmax(axis=1) - unique_classes = self.unique_classes - - pred_class = pred_class_indices.map_blocks( - lambda x: unique_classes[x], - meta=unique_classes[:0], - ) - if delayed: - return pred_class - else: - return pred_class.persist() - def predict_proba(self, X, delayed=True, **kwargs): """ Predicts the probability of each class for X. @@ -326,8 +287,6 @@ def predict_proba(self, X, delayed=True, **kwargs): ------- y : Dask cuDF dataframe or CuPy backed Dask Array (n_rows, n_classes) """ - if self._get_internal_model() is None: - self._set_internal_model(self._concat_treelite_models()) data = DistributedDataHandler.create(X, client=self.client) return self._predict_proba( X, delayed, output_collection_type=data.datatype, **kwargs diff --git a/python/cuml/cuml/dask/ensemble/randomforestregressor.py b/python/cuml/cuml/dask/ensemble/randomforestregressor.py index c4ccdea294..81c9f871d4 100755 --- a/python/cuml/cuml/dask/ensemble/randomforestregressor.py +++ b/python/cuml/cuml/dask/ensemble/randomforestregressor.py @@ -2,7 +2,7 @@ # SPDX-FileCopyrightText: Copyright (c) 2019-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 # -import dask.array +import warnings from cuml.dask.common.base import BaseEstimator, DelayedPredictionMixin from cuml.dask.ensemble.base import BaseRandomForestModel @@ -17,30 +17,13 @@ class RandomForestRegressor( regressors in an ensemble. This uses Dask to partition data over multiple GPUs (possibly on different nodes). - This implementation makes the following assumptions: - * The set of Dask workers used between instantiation, fit, - and predict are all consistent - * Training data comes in the form of cuDF dataframes or Dask Arrays - distributed so that each worker has at least one partition. - - The distributed algorithm uses an *embarrassingly-parallel* - approach. For a forest with `N` trees being built on `w` workers, each - worker simply builds `N/w` trees on the data it has available - locally. In many cases, partitioning the data so that each worker - builds trees on a subset of the total dataset works well, but - it generally requires the data to be well-shuffled in advance. - Alternatively, callers can replicate all of the data across - workers so that ``rf.fit`` receives `w` partitions, each containing the - same data. This would produce results approximately identical to - single-GPU fitting. - - Please check the single-GPU implementation of Random Forest - regressor for more information about the underlying algorithm. + During fitting, all workers that hold training rows collectively build the + same forest from the complete distributed dataset. Parameters ---------- n_estimators : int (default = 100) - total number of trees in the forest (not per-worker) + total number of trees in the forest split_criterion : int or string (default = ``2`` (``'mse'``)) The criterion used to split nodes.\n * ``0`` or ``'gini'`` for gini impurity @@ -93,20 +76,17 @@ class RandomForestRegressor( * If type ``float``, then ``min_samples_split`` represents a fraction and ``ceil(min_samples_split * n_rows)`` is the minimum number of samples for each split. - n_streams : int (default = 4 ) - Number of parallel streams used for forest building + n_streams : int + Deprecated. Distributed training currently builds trees serially to + preserve collective order. workers : optional, list of strings Dask addresses of workers to use for computation. If None, all available Dask workers will be used. random_state : int (default = None) Seed for the random number generator. Unseeded by default. - - ignore_empty_partitions: Boolean (default = False) - Specify behavior when a worker does not hold any data - while splitting. When True, it returns the results from workers - with data (the number of trained estimators will be less than - n_estimators) When False, throws a RuntimeError. - + ignore_empty_partitions: optional, boolean + Deprecated. This parameter no longer has any effect and + will be removed in release 26.12. """ def __init__( @@ -117,7 +97,7 @@ def __init__( verbose=False, n_estimators=100, random_state=None, - ignore_empty_partitions=False, + ignore_empty_partitions=None, **kwargs, ): super().__init__(client=client, verbose=verbose, **kwargs) @@ -138,12 +118,11 @@ def _construct_rf(n_estimators, random_state, **kwargs): n_estimators=n_estimators, random_state=random_state, **kwargs ) - def fit(self, X, y, broadcast_data=False): + def fit(self, X, y, broadcast_data=None): """ Fit the input data with a Random Forest regression model - IMPORTANT: X is expected to be partitioned with at least one partition - on each Dask worker being used by the forest (self.workers). + Only workers holding one or more training rows participate in fitting. When persisting data, you can use `cuml.dask.common.utils.persist_across_workers` to simplify this: @@ -175,16 +154,23 @@ def fit(self, X, y, broadcast_data=False): y : Dask cuDF DataFrame or CuPy backed Dask Array (n_rows, 1) Labels of training examples. **y must be partitioned the same way as X** - broadcast_data : bool, optional (default = False) - When set to True, the whole dataset is broadcasted - to train the workers, otherwise each worker - is trained on its partition + broadcast_data : bool, optional + Deprecated. This parameter no longer has effect and will + be removed in release 26.12. """ + if broadcast_data is not None: + warnings.warn( + ( + "broadcast_data parameter is no longer valid " + "and will be removed in release 26.12." + ), + FutureWarning, + stacklevel=2, + ) self.internal_model = None self._fit( model=self.rfs, dataset=(X, y), - broadcast_data=broadcast_data, ) return self @@ -195,7 +181,7 @@ def predict( default_chunk_size=None, align_bytes=None, delayed=True, - broadcast_data=False, + broadcast_data=None, ): """ Predicts the regressor outputs for X. @@ -206,7 +192,7 @@ def predict( Distributed dense matrix (floats or doubles) of shape (n_samples, n_features). layout : string (default = 'depth_first') - Specifies the in-memory layout of nodes in FIL forests. Options: + Specifies the in-memory layout of nodes in nvForest models. Options: 'depth_first', 'layered', 'breadth_first'. default_chunk_size : int, optional (default = None) Determines how batches are further subdivided for parallel processing. @@ -220,27 +206,24 @@ def predict( delayed : bool (default = True) Whether to do a lazy prediction (and return Delayed objects) or an eagerly executed one. - broadcast_data : bool (default = False) - If False, the trees are merged in a single model before the workers - perform inference on their share of the prediction workload. - When True, trees aren't merged. Instead each worker infers on the - whole prediction workload using its available trees. The results are - reduced on the client. May be advantageous when the model is larger - than the data used for inference. + broadcast_data : bool, optional + Deprecated. This parameter no longer has effect and will + be removed in release 26.12. Returns ------- y : Dask cuDF dataframe or CuPy backed Dask Array (n_rows, 1) """ - if broadcast_data: - return self.partial_inference( - X, - layout=layout, - default_chunk_size=default_chunk_size, - align_bytes=align_bytes, - delayed=delayed, + if broadcast_data is not None: + warnings.warn( + ( + "broadcast_data parameter is no longer valid " + "and will be removed in release 26.12." + ), + FutureWarning, + stacklevel=2, ) - return self._predict_using_fil( + return self._predict_using_nvforest( X, layout=layout, default_chunk_size=default_chunk_size, @@ -248,20 +231,6 @@ def predict( delayed=delayed, ) - def partial_inference(self, X, delayed, **kwargs): - partial_infs = self._partial_inference( - X=X, op_type="regression", delayed=delayed, **kwargs - ) - workers_weights = self._get_workers_weights() - merged_regressions = dask.array.average( - partial_infs, axis=1, weights=workers_weights - ) - - if delayed: - return merged_regressions - else: - return merged_regressions.persist() - def get_params(self, deep=True): """ Returns the value of all parameters diff --git a/python/cuml/cuml/ensemble/randomforest_common.pyx b/python/cuml/cuml/ensemble/randomforest_common.pyx index 502150b9fc..b283e5c098 100644 --- a/python/cuml/cuml/ensemble/randomforest_common.pyx +++ b/python/cuml/cuml/ensemble/randomforest_common.pyx @@ -37,6 +37,14 @@ from cuml.internals.treelite cimport ( ) +cdef extern from "cuml/ensemble/randomforest_mg_utils.hpp" namespace "ML::detail" nogil: + void cuml_rf_allreduce_validation_status( + const handle_t& handle, + const int* local_status, + int* global_status, + ) except + nogil + + cdef extern from "cuml/ensemble/randomforest.hpp" namespace "ML" nogil: cdef enum CRITERION: GINI, @@ -417,8 +425,11 @@ class BaseRandomForestModel(InteropMixin, Base): cdef uintptr_t sample_weight_ptr = ( 0 if sample_weight is None else sample_weight.data.ptr ) - cdef int n_rows = X.shape[0] - cdef int n_cols = X.shape[1] + cdef uint64_t n_rows = X.shape[0] + cdef uint64_t n_cols = X.shape[1] + cdef uint64_t parameter_n_rows = getattr( + self, "_distributed_n_rows", n_rows + ) cdef level_enum verbose = self._verbose_level cdef int n_classes = self.n_classes_ if is_classifier else 0 cdef bool input_row_major = not X.flags.f_contiguous @@ -469,19 +480,19 @@ class BaseRandomForestModel(InteropMixin, Base): ) cdef int min_samples_leaf = ( self.min_samples_leaf if isinstance(self.min_samples_leaf, int) - else math.ceil(self.min_samples_leaf * n_rows) + else math.ceil(self.min_samples_leaf * parameter_n_rows) ) cdef int min_samples_split = ( self.min_samples_split if isinstance(self.min_samples_split, int) - else max(2, math.ceil(self.min_samples_split * n_rows)) + else max(2, math.ceil(self.min_samples_split * parameter_n_rows)) ) cdef int n_bins - if self.n_bins > n_rows: + if self.n_bins > parameter_n_rows: warnings.warn("The number of bins, `n_bins` is greater than " "the number of samples used for training. " "Changing `n_bins` to number of training samples.") - n_bins = n_rows + n_bins = parameter_n_rows else: n_bins = self.n_bins @@ -503,7 +514,9 @@ class BaseRandomForestModel(InteropMixin, Base): ) cdef TreeliteModelHandle tl_handle - handle = get_handle(n_streams=n_streams_c) + handle = getattr(self, "_raft_handle", None) + if handle is None: + handle = get_handle(n_streams=n_streams_c) cdef handle_t* handle_ = handle.getHandle() # Store oob_score in C variable for nogil block @@ -604,7 +617,7 @@ class BaseRandomForestModel(InteropMixin, Base): TreeliteFreeModel(tl_handle), "Failed to free Treelite model:" ) - self._n_samples = y.shape[0] + self._n_samples = parameter_n_rows self._n_samples_bootstrap = ( self._n_samples if self.max_samples is None else max(round(self._n_samples * self.max_samples), 1) @@ -621,6 +634,29 @@ class BaseRandomForestModel(InteropMixin, Base): self.feature_importances_ = feature_importances return self + def _allreduce_validation_status(self, int local_status): + """Return whether input validation failed on any distributed rank.""" + handle = getattr(self, "_raft_handle", None) + if handle is None: + return local_status != 0 + + local_status_array = cp.asarray([local_status], dtype=cp.int32) + global_status_array = cp.empty_like(local_status_array) + cdef const int* local_status_ptr = ( + local_status_array.data.ptr + ) + cdef int* global_status_ptr = ( + global_status_array.data.ptr + ) + cdef handle_t* handle_ = handle.getHandle() + + with nogil: + cuml_rf_allreduce_validation_status( + handle_[0], local_status_ptr, global_status_ptr + ) + + return global_status_array.item() != 0 + def _get_inference_nvforest_model( self, layout="depth_first", diff --git a/python/cuml/cuml/ensemble/randomforestclassifier.py b/python/cuml/cuml/ensemble/randomforestclassifier.py index 53ce62ce7c..5b7af2b29a 100644 --- a/python/cuml/cuml/ensemble/randomforestclassifier.py +++ b/python/cuml/cuml/ensemble/randomforestclassifier.py @@ -259,6 +259,11 @@ def fit(self, X, y, sample_weight=None) -> "RandomForestClassifier": """ Perform Random Forest Classification on the input data """ + X, y, sample_weight = self._prepare_fit_inputs(X, y, sample_weight) + return self._fit_forest(X, y, sample_weight=sample_weight) + + def _prepare_fit_inputs(self, X, y, sample_weight=None): + classes = getattr(self, "_distributed_classes", True) X, y, sample_weight, classes = check_inputs( self, X, @@ -268,7 +273,7 @@ def fit(self, X, y, sample_weight=None) -> "RandomForestClassifier": order="A", y_dtype="int32", sample_weight_dtype="float64", - return_classes=True, + return_classes=classes, reset=True, ) self.classes_ = classes @@ -280,7 +285,7 @@ def fit(self, X, y, sample_weight=None) -> "RandomForestClassifier": sample_weight=sample_weight, dtype=np.float64, ) - return self._fit_forest(X, y, sample_weight=sample_weight) + return X, y, sample_weight @nvtx.annotate( message="predict RF-Classifier @randomforestclassifier.pyx", diff --git a/python/cuml/cuml/ensemble/randomforestregressor.py b/python/cuml/cuml/ensemble/randomforestregressor.py index 48043bf154..bae9f80d00 100644 --- a/python/cuml/cuml/ensemble/randomforestregressor.py +++ b/python/cuml/cuml/ensemble/randomforestregressor.py @@ -202,7 +202,11 @@ def fit(self, X, y, sample_weight=None) -> "RandomForestRegressor": Perform Random Forest Regression on the input data """ - X, y, sample_weight = check_inputs( + X, y, sample_weight = self._prepare_fit_inputs(X, y, sample_weight) + return self._fit_forest(X, y, sample_weight=sample_weight) + + def _prepare_fit_inputs(self, X, y, sample_weight=None): + return check_inputs( self, X, y, @@ -212,7 +216,6 @@ def fit(self, X, y, sample_weight=None) -> "RandomForestRegressor": sample_weight_dtype="float64", reset=True, ) - return self._fit_forest(X, y, sample_weight=sample_weight) @nvtx.annotate( message="predict RF-Regressor @randomforestclassifier.pyx", diff --git a/python/cuml/tests/dask/test_dask_random_forest.py b/python/cuml/tests/dask/test_dask_random_forest.py index 670d2b3532..d8d0c9db04 100644 --- a/python/cuml/tests/dask/test_dask_random_forest.py +++ b/python/cuml/tests/dask/test_dask_random_forest.py @@ -1,4 +1,4 @@ -# SPDX-FileCopyrightText: Copyright (c) 2019-2026, NVIDIA CORPORATION. +# SPDX-FileCopyrightText: Copyright (c) 2019-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 import json @@ -40,6 +40,10 @@ def _prep_training_data(c, X_train, y_train, partitions_per_worker): return X_train_df, y_train_df +def _get_treelite_bytes(model): + return model._treelite_model_bytes + + @pytest.mark.parametrize("partitions_per_worker", [3]) def test_rf_classification_multi_class(partitions_per_worker, cluster): # Use CUDA_VISIBLE_DEVICES to control the number of workers @@ -65,7 +69,7 @@ def test_rf_classification_multi_class(partitions_per_worker, cluster): ) cu_rf_params = { - "n_estimators": n_workers * 25, + "n_estimators": 25, "max_depth": 16, "n_bins": 256, "random_state": 10, @@ -75,7 +79,7 @@ def test_rf_classification_multi_class(partitions_per_worker, cluster): c, X_train, y_train, partitions_per_worker ) - cuml_mod = cuRFC_mg(**cu_rf_params, ignore_empty_partitions=True) + cuml_mod = cuRFC_mg(**cu_rf_params) cuml_mod.fit(X_train_df, y_train_df) X_test_dask_array = from_array(X_test) cuml_preds_gpu = cuml_mod.predict(X_test_dask_array).compute() @@ -101,7 +105,7 @@ def test_rf_classification_multi_class(partitions_per_worker, cluster): @pytest.mark.parametrize("dtype", [np.float32, np.float64]) @pytest.mark.parametrize("partitions_per_worker", [5]) -def test_rf_regression_dask_fil(partitions_per_worker, dtype, client): +def test_rf_regression_dask_nvforest(partitions_per_worker, dtype, client): n_workers = len(client.scheduler_info(n_workers=-1)["workers"]) # Use CUDA_VISIBLE_DEVICES to control the number of workers @@ -136,7 +140,7 @@ def test_rf_regression_dask_fil(partitions_per_worker, dtype, client): X_cudf_test = cudf.DataFrame(pd.DataFrame(X_test)) X_test_df = dask_cudf.from_cudf(X_cudf_test, npartitions=n_partitions) - cuml_mod = cuRFR_mg(**cu_rf_params, ignore_empty_partitions=True) + cuml_mod = cuRFR_mg(**cu_rf_params) cuml_mod.fit(X_train_df, y_train_df) cuml_mod_predict = cuml_mod.predict(X_test_df) @@ -147,6 +151,32 @@ def test_rf_regression_dask_fil(partitions_per_worker, dtype, client): assert acc_score >= 0.59 +def test_rf_regression_nan_on_one_worker(client): + workers = list(client.scheduler_info(n_workers=-1)["workers"]) + if len(workers) < 2: + pytest.skip("This test requires at least two workers") + + X_parts = [] + y_parts = [] + for rank, worker in enumerate(workers): + X_part = cudf.DataFrame( + np.arange(80, dtype=np.float32).reshape(20, 4) + rank + ) + y_part = cudf.Series(np.arange(20, dtype=np.float32) + rank) + if rank == 0: + X_part.iloc[0, 0] = np.nan + + X_parts.append(client.scatter(X_part, workers=[worker])) + y_parts.append(client.scatter(y_part, workers=[worker])) + + X = dask_cudf.from_delayed(X_parts, meta=X_part.iloc[:0]) + y = dask_cudf.from_delayed(y_parts, meta=y_part.iloc[:0]) + + model = cuRFR_mg(n_estimators=5, max_depth=3) + with pytest.raises(RuntimeError, match="Input X contains NaN"): + model.fit(X, y) + + @pytest.mark.parametrize("partitions_per_worker", [5]) def test_rf_classification_dask_array(partitions_per_worker, client): n_workers = len(client.scheduler_info(n_workers=-1)["workers"]) @@ -187,7 +217,7 @@ def test_rf_classification_dask_array(partitions_per_worker, client): @pytest.mark.parametrize("partitions_per_worker", [5]) -def test_rf_classification_dask_fil_predict_proba( +def test_rf_classification_dask_nvforest_predict_proba( partitions_per_worker, client ): n_workers = len(client.scheduler_info(n_workers=-1)["workers"]) @@ -210,7 +240,6 @@ def test_rf_classification_dask_fil_predict_proba( cu_rf_params = { "n_bins": 16, - "n_streams": 1, "n_estimators": 40, "max_depth": 16, } @@ -224,16 +253,18 @@ def test_rf_classification_dask_fil_predict_proba( cu_rf_mg = cuRFC_mg(**cu_rf_params) cu_rf_mg.fit(X_train_df, y_train_df) - fil_preds = cu_rf_mg.predict(X_test_df).compute() - fil_preds = fil_preds.to_numpy() - fil_preds_proba = cu_rf_mg.predict_proba(X_test_df).compute() - fil_preds_proba = fil_preds_proba.to_numpy() - np.testing.assert_equal(fil_preds, np.argmax(fil_preds_proba, axis=1)) + nvforest_preds = cu_rf_mg.predict(X_test_df).compute() + nvforest_preds = nvforest_preds.to_numpy() + nvforest_preds_proba = cu_rf_mg.predict_proba(X_test_df).compute() + nvforest_preds_proba = nvforest_preds_proba.to_numpy() + np.testing.assert_equal( + nvforest_preds, np.argmax(nvforest_preds_proba, axis=1) + ) - y_proba = np.zeros(np.shape(fil_preds_proba)) + y_proba = np.zeros(np.shape(nvforest_preds_proba)) y_proba[:, 1] = y_test y_proba[:, 0] = 1.0 - y_test - fil_mse = mean_squared_error(y_proba, fil_preds_proba) + nvforest_mse = mean_squared_error(y_proba, nvforest_preds_proba) sk_model = skrfc( n_estimators=cu_rf_params["n_estimators"], max_depth=cu_rf_params["max_depth"], @@ -245,11 +276,11 @@ def test_rf_classification_dask_fil_predict_proba( # The threshold is required as the test would intermitently # fail with a max difference of 0.029 between the two mse values - assert fil_mse <= sk_mse + 0.029 + assert nvforest_mse <= sk_mse + 0.029 @pytest.mark.parametrize("model_type", ["classification", "regression"]) -def test_rf_concatenation_dask(client, model_type): +def test_rf_distributed_model(client, model_type): n_workers = len(client.scheduler_info(n_workers=-1)["workers"]) X, y = make_classification( @@ -272,49 +303,60 @@ def test_rf_concatenation_dask(client, model_type): cu_rf_mg = cuRFR_mg(**cu_rf_params) cu_rf_mg.fit(X_df, y_df) - res1 = cu_rf_mg.predict(X_df) - res1.compute() - if cu_rf_mg.internal_model: - treelite_bytes = cu_rf_mg.internal_model._treelite_model_bytes - local_tl = treelite.Model.deserialize_bytes(treelite_bytes) - assert local_tl.num_tree == n_estimators + model = cu_rf_mg.get_combined_model() + treelite_bytes = model._treelite_model_bytes + local_tl = treelite.Model.deserialize_bytes(treelite_bytes) + assert local_tl.num_tree == n_estimators + worker_model_bytes = client.gather( + [ + client.submit(_get_treelite_bytes, model, workers=[worker]) + for worker, model in cu_rf_mg.rfs.items() + ] + ) + assert all(data == worker_model_bytes[0] for data in worker_model_bytes) + + +def test_rf_classification_uses_global_classes(client): + n_workers = len(client.scheduler_info(n_workers=-1)["workers"]) + if n_workers < 2: + pytest.skip("This test requires at least two workers") + + rows_per_worker = 100 + y = np.repeat(np.arange(n_workers), rows_per_worker).astype(np.int32) + X = np.column_stack((y, np.arange(y.size))).astype(np.float32) + X_dask, y_dask = _prep_training_data(client, X, y, partitions_per_worker=1) + model = cuRFC_mg( + n_estimators=1, + bootstrap=False, + max_depth=4, + n_bins=max(2, n_workers), + random_state=42, + ).fit(X_dask, y_dask) -@pytest.mark.parametrize("ignore_empty_partitions", [True, False]) -def test_single_input_regression(client, ignore_empty_partitions): + np.testing.assert_array_equal( + model.get_combined_model().classes_, np.arange(n_workers) + ) + + +def test_single_input_regression(client): X, y = make_classification(n_samples=1, n_classes=1) X = X.astype(np.float32) y = y.astype(np.float32) X, y = _prep_training_data(client, X, y, partitions_per_worker=2) - cu_rf_mg = cuRFR_mg( - n_bins=1, - ignore_empty_partitions=ignore_empty_partitions, - ) - - if ( - ignore_empty_partitions - or len(client.scheduler_info(n_workers=-1)["workers"].keys()) == 1 - ): - cu_rf_mg.fit(X, y) - cuml_mod_predict = cu_rf_mg.predict(X) - cuml_mod_predict = cp.asnumpy(cp.array(cuml_mod_predict.compute())) - y = cp.asnumpy(cp.array(y.compute())) - assert y[0] == cuml_mod_predict[0] - - else: - with pytest.raises(ValueError): - cu_rf_mg.fit(X, y) + cu_rf_mg = cuRFR_mg(n_bins=1) + cu_rf_mg.fit(X, y) + cuml_mod_predict = cu_rf_mg.predict(X) + cuml_mod_predict = cp.asnumpy(cp.array(cuml_mod_predict.compute())) + y = cp.asnumpy(cp.array(y.compute())) + assert y[0] == cuml_mod_predict[0] @pytest.mark.parametrize("max_depth", [1, 2, 3, 5, 10, 15, 20]) -@pytest.mark.parametrize("n_estimators", [5, 10, 20]) +@pytest.mark.parametrize("n_estimators", [1, 5, 10, 20]) def test_rf_data_count(client, max_depth, n_estimators): n_workers = len(client.scheduler_info(n_workers=-1)["workers"]) - if n_estimators < n_workers: - err_msg = "n_estimators cannot be lower than number of dask workers" - pytest.xfail(err_msg) - n_samples_per_worker = 350 X, y = make_classification( @@ -333,7 +375,6 @@ def test_rf_data_count(client, max_depth, n_estimators): split_criterion=0, min_samples_leaf=2, random_state=23707, - n_streams=1, n_estimators=n_estimators, max_leaves=-1, max_depth=max_depth, @@ -355,8 +396,8 @@ def check_count(node, nodes): for tree in json_obj["trees"]: nodes = tree["nodes"] - # The root's count should be equal to the number of rows in the data - assert nodes[0]["data_count"] == n_samples_per_worker + # The root contains rows from the complete distributed dataset. + assert nodes[0]["data_count"] == n_samples_per_worker * n_workers # Check that the data_count accumulates properly as you move up the tree for node in nodes: check_count(node, nodes) @@ -371,7 +412,7 @@ def test_unlimited_max_depth_classifier(client): y = y.astype(np.int32) X_dask, y_dask = _prep_training_data(client, X, y, partitions_per_worker=1) - clf = cuRFC_mg(n_estimators=n_workers * 5, max_depth=None) + clf = cuRFC_mg(n_estimators=5, max_depth=None) clf.fit(X_dask, y_dask) preds = cp.asnumpy(cp.array(clf.predict(X_dask).compute())) assert len(preds) == len(y) @@ -386,22 +427,17 @@ def test_unlimited_max_depth_regressor(client): y = y.astype(np.float32) X_dask, y_dask = _prep_training_data(client, X, y, partitions_per_worker=1) - reg = cuRFR_mg(n_estimators=n_workers * 5, max_depth=None) + reg = cuRFR_mg(n_estimators=5, max_depth=None) reg.fit(X_dask, y_dask) preds = cp.asnumpy(cp.array(reg.predict(X_dask).compute())) assert len(preds) == len(y) @pytest.mark.parametrize("estimator_type", ["regression", "classification"]) -def test_rf_get_combined_model_right_aftter_fit(client, estimator_type): +def test_rf_get_model_right_after_fit(client, estimator_type): max_depth = 3 n_estimators = 5 - n_workers = len(client.scheduler_info(n_workers=-1)["workers"]) - if n_estimators < n_workers: - err_msg = "n_estimators cannot be lower than number of dask workers" - pytest.xfail(err_msg) - X, y = make_classification() X = X.astype(np.float32) if estimator_type == "classification": @@ -409,7 +445,6 @@ def test_rf_get_combined_model_right_aftter_fit(client, estimator_type): max_features=1.0, max_samples=1.0, n_bins=16, - n_streams=1, n_estimators=n_estimators, max_leaves=-1, max_depth=max_depth, @@ -420,7 +455,6 @@ def test_rf_get_combined_model_right_aftter_fit(client, estimator_type): max_features=1.0, max_samples=1.0, n_bins=16, - n_streams=1, n_estimators=n_estimators, max_leaves=-1, max_depth=max_depth, @@ -437,78 +471,3 @@ def test_rf_get_combined_model_right_aftter_fit(client, estimator_type): assert isinstance(single_gpu_model, cuRFR_sg) else: assert False - - -@pytest.mark.parametrize("model_type", ["classification", "regression"]) -@pytest.mark.parametrize("fit_broadcast", [True, False]) -@pytest.mark.parametrize("transform_broadcast", [True, False]) -def test_rf_broadcast(model_type, fit_broadcast, transform_broadcast, client): - # Use CUDA_VISIBLE_DEVICES to control the number of workers - workers = list(client.scheduler_info(n_workers=-1)["workers"].keys()) - n_workers = len(workers) - - if model_type == "classification": - X, y = make_classification( - n_samples=n_workers * 10000, - n_features=20, - n_informative=15, - n_classes=4, - n_clusters_per_class=1, - random_state=999, - ) - y = y.astype(np.int32) - else: - X, y = make_regression( - n_samples=n_workers * 10000, - n_features=20, - n_informative=5, - random_state=123, - ) - y = y.astype(np.float32) - X = X.astype(np.float32) - - X_train, X_test, y_train, y_test = train_test_split( - X, y, test_size=n_workers * 100, random_state=123 - ) - - X_train_df, y_train_df = _prep_training_data(client, X_train, y_train, 1) - X_test_dask_array = from_array(X_test) - - n_estimators = n_workers * 8 - - if model_type == "classification": - cuml_mod = cuRFC_mg( - n_estimators=n_estimators, - max_depth=8, - n_bins=16, - ignore_empty_partitions=True, - ) - cuml_mod.fit(X_train_df, y_train_df, broadcast_data=fit_broadcast) - cuml_mod_predict = cuml_mod.predict( - X_test_dask_array, broadcast_data=transform_broadcast - ) - - cuml_mod_predict = cuml_mod_predict.compute() - cuml_mod_predict = cp.asnumpy(cuml_mod_predict) - acc_score = accuracy_score(cuml_mod_predict, y_test, normalize=True) - assert acc_score >= 0.68 - - else: - cuml_mod = cuRFR_mg( - n_estimators=n_estimators, - max_depth=8, - n_bins=16, - ignore_empty_partitions=True, - ) - cuml_mod.fit(X_train_df, y_train_df, broadcast_data=fit_broadcast) - cuml_mod_predict = cuml_mod.predict( - X_test_dask_array, broadcast_data=transform_broadcast - ) - - cuml_mod_predict = cuml_mod_predict.compute() - cuml_mod_predict = cp.asnumpy(cuml_mod_predict) - acc_score = r2_score(y_test, cuml_mod_predict) - assert acc_score >= 0.72 - - if transform_broadcast: - assert cuml_mod.internal_model is None