code: def _sample_map_partitions(population, k):
      """
      Reservoir sampling strategy based on the L algorithm
      See https://en.wikipedia.org/wiki/Reservoir_sampling#An_optimal_algorithm
      """
  
      reservoir, stream_length = [], 0
      stream = iter(population)
      for e in islice(stream, k):
          reservoir.append(e)
          stream_length += 1
  
      w = math.exp(math.log(rnd.random()) / k)
      nxt = (k - 1) + _geometric(w)
  
      for i, e in enumerate(stream, k):
          if i == nxt:
              reservoir[rnd.randrange(k)] = e
              w *= math.exp(math.log(rnd.random()) / k)
              nxt += _geometric(w)
          stream_length += 1
  
      return reservoir, stream_length
  
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 7 for calling a global function with example usage: _sample_map_partitions(population, k) and returns: reservoir,  stream_length