code: def _prep_training_data(
      c, X_train, partitions_per_worker, reverse_order=False
  ):
      workers = c.has_what().keys()
  
      if reverse_order:
          workers = list(workers)[::-1]
  
      n_partitions = partitions_per_worker * len(workers)
  
      X_cudf = cudf.DataFrame.from_pandas(pd.DataFrame(X_train))
  
      X_train_df = dask_cudf.from_cudf(X_cudf, npartitions=n_partitions)
      (X_train_df,) = dask_utils.persist_across_workers(
          c, [X_train_df], workers=list(workers)
      )
  
      return X_train_df
  
file_path: cuml/python/cuml/tests/dask/test_dask_nearest_neighbors.py
imports: import platform
  from cuml.testing.utils import array_equal
  from sklearn.neighbors import KNeighborsClassifier
  from cuml.testing.utils import unit_param, quality_param, stress_param
  from cuml.dask.common import utils
  from cuml.common import has_scipy
  from cuml.internals.safe_imports import cpu_only_import
  import pytest
  from cuml.internals.safe_imports import gpu_only_import
label_desc: Use image node_id 2 for calling a global function with example usage: _prep_training_data(c, X_train, partitions_per_worker, reverse_order) and returns: X_train_df