code: def fit(self, X, y, **kwargs):
      X = self._check_array(X)
      estimatord = dask.delayed(self.estimator)
  
      Xs = X.to_delayed()
      ys = y.to_delayed()
      if isinstance(X, da.Array):
          Xs = Xs.flatten()
      if isinstance(y, da.Array):
          ys = ys.flatten()
  
      if len(Xs) != len(ys):
          raise ValueError(
              f"The number of blocks in X and y must match. {len(Xs)} != {len(ys)}"
          )
  
      estimators = [
          dask.delayed(sklearn.base.clone)(estimatord)
          for _ in range(len(Xs))
      ]
      results = [
          estimator_.fit(X_, y_, **kwargs)
          for estimator_, X_, y_, in zip(estimators, Xs, ys)
      ]
      results = list(dask.compute(*results))
      self.estimators_ = results
  
imports: import dask
  import dask.array
  import dask.dataframe
  import numpy
  import sklearn.base
  from sklearn.utils.validation import check_is_fitted
  from ..base import ClassifierMixin, RegressorMixin
  from ..utils import check_array
label_desc: Use image node_id 3 for calling the BlockwiseBase obj's underlying member method code with example usage: obj.fit(X, y) without return types