code: def _dask_maamped(
      dask_client,
      T_A,
      T_B,
      m,
      excl_zone,
      T_A_subseq_isfinite,
      T_B_subseq_isfinite,
      p,
      include,
      discords,
  ):
      """
      Compute the multi-dimensional non-normalized (i.e., without z-normalization) matrix
      profile with a distributed dask cluster
  
      This is a highly distributed implementation around the Numba JIT-compiled
      parallelized `_maamp` function which computes the multi-dimensional matrix
      profile according to STOMP. Note that only self-joins are supported.
  
      Parameters
      ----------
      dask_client : client
          A Dask Distributed client. Setting up a distributed cluster is beyond
          the scope of this library. Please refer to the Dask Distributed
          documentation.
  
      T_A : numpy.ndarray
          The time series or sequence for which to compute the multi-dimensional
          matrix profile. Each row in `T_A` represents data from a different
          dimension while each column in `T_A` represents data from the same
          dimension.
  
      T_B : numpy.ndarray
          The time series or sequence that will be used to annotate T_A. For every
          subsequence in T_A, its nearest neighbor in T_B will be recorded. Default is
          `None` which corresponds to a self-join.
  
      m : int
          Window size
  
      excl_zone : int
          The half width for the exclusion zone relative to the current
          sliding window
  
      T_A_subseq_isfinite : numpy.ndarray
          A boolean array that indicates whether a subsequence in `T_A` contains a
          `np.nan`/`np.inf` value (False)
  
      T_B_subseq_isfinite : numpy.ndarray
          A boolean array that indicates whether a subsequence in `T_B` contains a
          `np.nan`/`np.inf` value (False)
  
      p : float
          The p-norm to apply for computing the Minkowski distance. Minkowski distance is
          typically used with `p` being 1 or 2, which correspond to the Manhattan distance
          and the Euclidean distance, respectively.
  
      include : numpy.ndarray
          A list of (zero-based) indices corresponding to the dimensions in `T` that
          must be included in the constrained multidimensional motif search.
          For more information, see Section IV D in:
  
          `DOI: 10.1109/ICDM.2017.66 \
          <https://www.cs.ucr.edu/~eamonn/Motif_Discovery_ICDM.pdf>`__
  
      discords : bool
          When set to `True`, this reverses the distance profile to favor discords rather
          than motifs. Note that indices in `include` are still maintained and respected.
  
      Returns
      -------
      P : numpy.ndarray
          The multi-dimensional matrix profile. Each row of the array corresponds
          to each matrix profile for a given dimension (i.e., the first row is
          the 1-D matrix profile and the second row is the 2-D matrix profile).
  
      I : numpy.ndarray
          The multi-dimensional matrix profile index where each row of the array
          corresponds to each matrix profile index for a given dimension.
      """
      d, n = T_B.shape
      l = n - m + 1
      P = np.empty((d, l), dtype=np.float64)
      I = np.empty((d, l), dtype=np.int64)
  
      hosts = list(dask_client.ncores().keys())
      nworkers = len(hosts)
  
      step = 1 + l // nworkers
  
      for i, start in enumerate(range(0, l, step)):
          P[:, start], I[:, start] = _get_first_maamp_profile(
              start,
              T_A,
              T_B,
              m,
              excl_zone,
              T_B_subseq_isfinite,
              p,
              include,
              discords,
          )
      # Scatter data to Dask cluster
      T_A_future = dask_client.scatter(T_A, broadcast=True, hash=False)
      T_A_subseq_isfinite_future = dask_client.scatter(
          T_A_subseq_isfinite, broadcast=True, hash=False
      )
      T_B_subseq_isfinite_future = dask_client.scatter(
          T_B_subseq_isfinite, broadcast=True, hash=False
      )
  
      p_norm_futures = []
      p_norm_first_futures = []
  
      for i, start in enumerate(range(0, l, step)):
          p_norm, p_norm_first = _get_multi_p_norm(start, T_A, m)
  
          p_norm_future = dask_client.scatter(
              p_norm, workers=[hosts[i]], hash=False
          )
          p_norm_first_future = dask_client.scatter(
              p_norm_first, workers=[hosts[i]], hash=False
          )
  
          p_norm_futures.append(p_norm_future)
          p_norm_first_futures.append(p_norm_first_future)
  
      futures = []
      for i, start in enumerate(range(0, l, step)):
          stop = min(l, start + step)
  
          futures.append(
              dask_client.submit(
                  _maamp,
                  T_A_future,
                  m,
                  stop,
                  excl_zone,
                  T_A_subseq_isfinite_future,
                  T_B_subseq_isfinite_future,
                  p,
                  p_norm_futures[i],
                  p_norm_first_futures[i],
                  l,
                  start + 1,
                  include,
                  discords,
              )
          )
  
      results = dask_client.gather(futures)
      for i, start in enumerate(range(0, l, step)):
          stop = min(l, start + step)
          P[:, start + 1 : stop], I[:, start + 1 : stop] = results[i]
  
      return P, I
  
imports: import numpy
  from .None import config, core
  from .maamp import _get_first_maamp_profile, _get_multi_p_norm, _maamp
label_desc: Use image node_id 1 for calling a global function with example usage: _dask_maamped(dask_client, T_A, T_B, m, excl_zone, T_A_subseq_isfinite, T_B_subseq_isfinite, p, include, discords) and returns: P,  I