code: def get(
      dsk: Mapping,
      keys: Sequence[Key] | Key,
      num_workers=None,
      func_loads=None,
      func_dumps=None,
      optimize_graph=True,
      pool=None,
      initializer=None,
      chunksize=None,
      **kwargs,
  ):
      """Multiprocessed get function appropriate for Bags
  
      Parameters
      ----------
      dsk : dict
          dask graph
      keys : object or list
          Desired results from graph
      num_workers : int
          Number of worker processes (defaults to number of cores)
      func_dumps : function
          Function to use for function serialization (defaults to cloudpickle.dumps)
      func_loads : function
          Function to use for function deserialization (defaults to cloudpickle.loads)
      optimize_graph : bool
          If True [default], `fuse` is applied to the graph before computation.
      pool : Executor or Pool
          Some sort of `Executor` or `Pool` to use
      initializer: function
          Ignored if ``pool`` has been set.
          Function to initialize a worker process before running any tasks in it.
      chunksize: int, optional
          Size of chunks to use when dispatching work.
          Defaults to 5 as some batching is helpful.
          If -1, will be computed to evenly divide ready work across workers.
      """
      chunksize = chunksize or config.get("chunksize", 6)
      pool = pool or config.get("pool", None)
      initializer = initializer or config.get(
          "multiprocessing.initializer", None
      )
      num_workers = (
          num_workers or config.get("num_workers", None) or CPU_COUNT
      )
      if pool is None:
          # In order to get consistent hashing in subprocesses, we need to set a
          # consistent seed for the Python hash algorithm. Unfortunately, there
          # is no way to specify environment variables only for the Pool
          # processes, so we have to rely on environment variables being
          # inherited.
          if os.environ.get("PYTHONHASHSEED") in (None, "0"):
              # This number is arbitrary; it was chosen to commemorate
              # https://github.com/dask/dask/issues/6640.
              os.environ["PYTHONHASHSEED"] = "6640"
          context = get_context()
          initializer = partial(
              initialize_worker_process, user_initializer=initializer
          )
          pool = ProcessPoolExecutor(
              num_workers, mp_context=context, initializer=initializer
          )
          cleanup = True
      else:
          if initializer is not None:
              warn(
                  "The ``initializer`` argument is ignored when ``pool`` is provided. "
                  "The user should configure ``pool`` with the needed ``initializer`` "
                  "on creation."
              )
          if isinstance(pool, multiprocessing.pool.Pool):
              pool = MultiprocessingPoolExecutor(pool)
          cleanup = False
  
      # Optimize Dask
      dsk = ensure_dict(dsk)
      dsk2, dependencies = cull(dsk, keys)
      if optimize_graph:
          dsk3, dependencies = fuse(dsk2, keys, dependencies)
      else:
          dsk3 = dsk2
  
      # We specify marshalling functions in order to catch serialization
      # errors and report them to the user.
      loads = func_loads or config.get("func_loads", None) or _loads
      dumps = func_dumps or config.get("func_dumps", None) or _dumps
  
      # Note former versions used a multiprocessing Manager to share
      # a Queue between parent and workers, but this is fragile on Windows
      # (issue #1652).
      try:
          # Run
          result = get_async(
              pool.submit,
              pool._max_workers,
              dsk3,
              keys,
              get_id=_process_get_id,
              dumps=dumps,
              loads=loads,
              pack_exception=pack_exception,
              raise_exception=reraise,
              chunksize=chunksize,
              **kwargs,
          )
      finally:
          if cleanup:
              pool.shutdown()
      return result
  
imports: from __future__ import annotations
  import copyreg
  import multiprocessing
  import multiprocessing.pool
  import os
  import pickle
  import sys
  import traceback
  from collections.abc import Mapping, Sequence
  from concurrent.futures import ProcessPoolExecutor
  from functools import partial
  from warnings import warn
  import cloudpickle
  from dask import config
  from dask.local import MultiprocessingPoolExecutor, get_async, reraise
  from dask.optimization import cull, fuse
  from dask.system import CPU_COUNT
  from dask.typing import Key
  from dask.utils import ensure_dict
label_desc: Use image node_id 6 for calling a global function with example usage: get(dsk, keys, num_workers, func_loads, func_dumps, optimize_graph, pool, initializer, chunksize) and returns: result