code: def init_scalable(
      X,
      n_clusters,
      random_state=None,
      max_iter=None,
      oversampling_factor=2,
      n_init="auto",
  ):
      """K-Means initialization using k-means||
  
      This is algorithm 2 in Scalable K-Means++ (2012).
      """
  
      logger.info("Initializing with k-means||")
      # Step 1: Initialize Centers
      idx = 0
      centers = da.compute(X[idx, np.newaxis])[0]
      c_idx = {idx}
  
      # Step 2: Initialize cost
      (cost,) = compute(evaluate_cost(X, centers))
  
      if cost == 0:
          n_iter = 0
      else:
          n_iter = int(np.round(np.log(cost)))
  
      if max_iter is not None:
          n_iter = min(max_iter, n_iter)
  
      # Steps 3 - 6: update candidate Centers
      for i in range(n_iter):
          with _timer(
              "init iteration %2d/%2d , %2d centers"
              % (i + 1, n_iter, len(c_idx)),
              _logger=logger,
          ):
              new_idxs = _sample_points(
                  X, centers, oversampling_factor, random_state
              )
              new_idxs = set(*compute(new_idxs))
              c_idx |= new_idxs
  
          # Sort before slicing, for better performance / memory
          # usage with the scheduler.
          # See https://github.com/dask/dask-ml/issues/39
          centers = X[sorted(c_idx)].compute()
  
      # XXX: scikit-learn doesn't have weighted k-means.
      # The paper weights each center by the number of points closest to it.
      # https://stackoverflow.com/a/37198799/1889400 claims you can scale the
      # features before clustering, but that doesn't seem right.
      # I think that replicating the *points*, proportional to the number of
      # original points closest to the candidate centers, would be a better way
      # to do that.
  
      if len(centers) < n_clusters:
          logger.warning(
              "Found fewer than %d clusters in init (found %d).",
              n_clusters,
              len(centers),
          )
          # supplement with random
          need = n_clusters - len(centers)
          locs = sorted(
              random_state.choice(
                  np.arange(0, len(X)),
                  size=need,
                  replace=False,
                  chunks=len(X),
              )
          )
          extra = X[locs].compute()
          return np.vstack([centers, extra])
      else:
          # Step 7, 8 without weights
          # dask RandomState objects aren't valid for scikit-learn
          rng2 = (
              random_state.randint(0, np.iinfo("i4").max - 1, chunks=())
              .compute(scheduler="single-threaded")
              .item()
          )
          km = sklearn.cluster.KMeans(
              n_clusters, random_state=rng2, n_init=n_init
          )
          km.fit(centers)
  
      return km.cluster_centers_
  
imports: import logging
  from multiprocessing import cpu_count
  from numbers import Integral
  import dask.array
  import dask.dataframe
  import numpy
  import pandas
  import sklearn.cluster
  from dask import compute
  from sklearn.base import BaseEstimator, TransformerMixin
  from sklearn.utils.extmath import squared_norm
  from sklearn.utils.validation import check_is_fitted
  from .._compat import blockwise
  from .._utils import draw_seed
  from ..metrics import euclidean_distances, pairwise_distances, pairwise_distances_argmin_min
  from ..utils import _timed, _timer, check_array, row_norms
  from ._compat import _kmeans_plusplus
  import numba
label_desc: Use image node_id 6 for calling a global function with example usage: init_scalable(X, n_clusters, random_state, max_iter, oversampling_factor, n_init) and returns: km, np