code: def _prep_training_data_sparse(
      c, X_train, y_train, partitions_per_worker
  ):
      "The implementation follows test_dask_tfidf.create_cp_sparse_dask_array"
      import dask.array as da
  
      workers = c.has_what().keys()
      target_n_partitions = partitions_per_worker * len(workers)
  
      def cal_chunks(dataset, n_partitions):
          n_samples = dataset.shape[0]
          n_samples_per_part = int(n_samples / n_partitions)
          chunk_sizes = [n_samples_per_part] * n_partitions
          samples_last_row = n_samples - (
              (n_partitions - 1) * n_samples_per_part
          )
          chunk_sizes[-1] = samples_last_row
          return tuple(chunk_sizes)
  
      assert (
          X_train.shape[0] == y_train.shape[0]
      ), "the number of data records is not equal to the number of labels"
      target_chunk_sizes = cal_chunks(X_train, target_n_partitions)
  
      X_da = da.from_array(X_train, chunks=(target_chunk_sizes, -1))
      y_da = da.from_array(y_train, chunks=target_chunk_sizes)
  
      X_da, y_da = dask_utils.persist_across_workers(
          c, [X_da, y_da], workers=workers
      )
      return X_da, y_da
  
imports: from cuml.internals.safe_imports import gpu_only_import
  import pytest
  from cuml.dask.common import utils
  from functools import partial
  from sklearn.metrics import accuracy_score, mean_squared_error
  from sklearn.datasets import make_classification
  from sklearn.linear_model import LogisticRegression
  from cuml.internals.safe_imports import cpu_only_import
  from cuml.testing.utils import array_equal
  from scipy.sparse import csr_matrix
label_desc: Use image node_id 2 for calling a global function with example usage: _prep_training_data_sparse(c, X_train, y_train, partitions_per_worker) and returns: X_da,  y_da, tuple