code: def make_blobs(
      n_samples=100,
      n_features=2,
      centers=None,
      cluster_std=1.0,
      center_box=(-10.0, 10.0),
      shuffle=True,
      random_state=None,
      chunks=None,
  ):
      """
      Generate isotropic Gaussian blobs for clustering.
  
      This can be used to generate very large Dask arrays on a cluster of
      machines. When using Dask in distributed mode, the client machine
      only needs to allocate a single block's worth of data.
  
      Parameters
      ----------
      n_samples : int or array-like, optional (default=100)
          If int, it is the total number of points equally divided among
          clusters.
          If array-like, each element of the sequence indicates
          the number of samples per cluster.
  
      n_features : int, optional (default=2)
          The number of features for each sample.
  
      centers : int or array of shape [n_centers, n_features], optional
          (default=None)
          The number of centers to generate, or the fixed center locations.
          If n_samples is an int and centers is None, 3 centers are generated.
          If n_samples is array-like, centers must be
          either None or an array of length equal to the length of n_samples.
  
      cluster_std : float or sequence of floats, optional (default=1.0)
          The standard deviation of the clusters.
  
      center_box : pair of floats (min, max), optional (default=(-10.0, 10.0))
          The bounding box for each cluster center when centers are
          generated at random.
  
      shuffle : boolean, optional (default=True)
          Shuffle the samples.
  
      random_state : int, RandomState instance or None (default)
          Determines random number generation for dataset creation. Pass an int
          for reproducible output across multiple function calls.
          See :term:`Glossary <random_state>`.
  
      chunks : int, tuple
          How to chunk the array. Must be one of the following forms:
          -   A blocksize like 1000.
          -   A blockshape like (1000, 1000).
          -   Explicit sizes of all blocks along all dimensions like
              ((1000, 1000, 500), (400, 400)).
  
      Returns
      -------
      X : array of shape [n_samples, n_features]
          The generated samples.
  
      y : array of shape [n_samples]
          The integer labels for cluster membership of each sample.
  
      Examples
      --------
      >>> from dask_ml.datasets import make_blobs
      >>> X, y = make_blobs(n_samples=100000, chunks=10000)
      >>> X
      dask.array<..., shape=(100000, 2), dtype=float64, chunksize=(10000, 2)>
      >>> y
      dask.array<concatenate, shape=(100000,), dtype=int64, chunksize=(10000,)>
  
      See Also
      --------
      make_classification: a more intricate variant
      """
      chunks = da.core.normalize_chunks(chunks, (n_samples, n_features))
      _check_axis_partitioning(chunks, n_features)
  
      if centers is None:
          # TODO: non-int n_samples?
          centers = 3
      if isinstance(centers, numbers.Integral):
          # Make a prototype
          n_centers = centers
          X, y = sklearn.datasets.make_blobs(
              n_samples=chunks[0][0],
              n_features=n_features,
              centers=centers,
              shuffle=shuffle,
              cluster_std=cluster_std,
              center_box=center_box,
              random_state=random_state,
          )
          centers = []
          centers = np.zeros((n_centers, n_features))
  
          for i in range(n_centers):
              centers[i] = X[y == i].mean(0)
  
      objs = [
          dask.delayed(sklearn.datasets.make_blobs, nout=2)(
              n_samples=n_samples_per_block,
              n_features=n_features,
              centers=centers,
              cluster_std=cluster_std,
              shuffle=shuffle,
              center_box=center_box,
              random_state=i,
          )
          for i, n_samples_per_block in enumerate(chunks[0])
      ]
      Xobjs, yobjs = zip(*objs)
  
      Xarrs = [
          da.from_delayed(arr, shape=(n, n_features), dtype="f8")
          for arr, n in zip(Xobjs, chunks[0])
      ]
      X_big = da.vstack(Xarrs)
  
      yarrs = [
          da.from_delayed(arr, shape=(n,), dtype=np.dtype("int"))
          for arr, n in zip(yobjs, chunks[0])
      ]
      y_big = da.hstack(yarrs)
      return X_big, y_big
  
file_path: dask-ml/dask_ml/datasets.py
imports: import numbers
  from datetime import timedelta
  import dask
  import dask.array
  import dask.dataframe
  import numpy
  import sklearn.datasets
  import sklearn.utils
  import dask_ml.utils
label_desc: Use image node_id 3 for calling a global function with example usage: make_blobs(n_samples, n_features, centers, cluster_std, center_box, shuffle, random_state, chunks) and returns: X_big,  y_big