code: def _sample_reduce(reduce_iter, k, replace):
      """
      Reduce function used on the sample and choice functions.
  
      Parameters
      ----------
      reduce_iter : iterable
          Each element is a tuple coming generated by the _sample_map_partitions function.
      replace: bool
          If True, sample with replacement. If False, sample without replacement.
  
      Returns a sequence of uniformly distributed samples;
      """
      ns_ks = []
      s = []
      n = 0
      # unfolding reduce outputs
      for i in reduce_iter:
          (s_i, n_i) = i
          s.extend(s_i)
          n += n_i
          k_i = len(s_i)
          ns_ks.append((n_i, k_i))
  
      if k > n and not replace:
          return s, n
  
      # creating the probability array
      p = []
      for n_i, k_i in ns_ks:
          if k_i > 0:
              p_i = n_i / (k_i * n)
              p += [p_i] * k_i
  
      sample_func = (
          rnd.choices
          if replace
          else _weighted_sampling_without_replacement
      )
      return sample_func(population=s, weights=p, k=k), n
  
file_path: dask/dask/bag/random.py
imports: from __future__ import annotations
  import heapq
  import math
  import random
  from functools import partial
  from itertools import islice
  from dask.bag.core import Bag
label_desc: Use image node_id 3 for calling a global function with example usage: _sample_reduce(reduce_iter, k, replace) and returns: sample_func,  n, s,  n