code: def create_merge_tree(func, keys, token, level=0):
      """Create a task tree that merges all the keys with a reduction function.
  
      Parameters
      ----------
      func: callable
          Reduction function that accepts a single list of values to reduce.
      keys: iterable
          Keys to reduce from the source dask graph.
      token: object
          Included in each key of the returned dict.
      level: int, default 0
          The token-level to begin with.
  
      This creates a k-ary tree where k depends on the current level and is
      greater the further away a node is from the root node.  This reduces the
      total number of nodes (thereby reducing scheduler overhead), but still
      has beneficial properties of trees.
  
      For reasonable numbers of keys, N < 1e5, the total number of nodes in the
      tree is roughly ``N**0.78``.  For 1e5 < N < 2e5, is it roughly ``N**0.8``.
      """
      prev_width = len(keys)
      prev_keys = iter(keys)
      rv = {}
      while prev_width > 1:
          width = tree_width(prev_width)
          groups = tree_groups(prev_width, width)
          keys = [(token, level, i) for i in range(width)]
  
          for num, key in zip(groups, keys):
              rv[key] = (func, list(take(num, prev_keys)))
  
          prev_width = width
          prev_keys = iter(keys)
          level += 1
      return rv
  
imports: from __future__ import annotations
  import math
  import numpy
  import pandas
  from pandas.api.types import is_datetime64_dtype, is_integer_dtype
  from tlz import merge, merge_sorted, take
  from dask.base import tokenize
  from dask.dataframe.core import Series
  from dask.dataframe.dispatch import tolist_dispatch
  from dask.utils import is_cupy_type, random_state_data
label_desc: Use image node_id 4 for calling a global function with example usage: create_merge_tree(func, keys, token, level) and returns: rv