code: def optimize(dsk, keys, **kwargs):
      if not isinstance(keys, (list, set)):
          keys = [keys]
      keys = list(core.flatten(keys))
  
      if not isinstance(dsk, HighLevelGraph):
          dsk = HighLevelGraph.from_collections(
              id(dsk), dsk, dependencies=()
          )
      else:
          # Perform Blockwise optimizations for HLG input
          dsk = optimize_dataframe_getitem(dsk, keys=keys)
          dsk = optimize_blockwise(dsk, keys=keys)
          dsk = fuse_roots(dsk, keys=keys)
      dsk = dsk.cull(set(keys))
  
      # Do not perform low-level fusion unless the user has
      # specified True explicitly. The configuration will
      # be None by default.
      if not config.get("optimization.fuse.active"):
          return dsk
  
      dependencies = dsk.get_all_dependencies()
      dsk = ensure_dict(dsk)
  
      fuse_subgraphs = config.get("optimization.fuse.subgraphs")
      if fuse_subgraphs is None:
          fuse_subgraphs = True
      dsk, _ = fuse(
          dsk,
          keys,
          dependencies=dependencies,
          fuse_subgraphs=fuse_subgraphs,
      )
      dsk, _ = cull(dsk, keys)
      return dsk
  
file_path: dask/dask/dataframe/optimize.py
imports: from __future__ import annotations
  import operator
  import numpy
  from dask import config, core
  from dask.blockwise import Blockwise, fuse_roots, optimize_blockwise
  from dask.highlevelgraph import HighLevelGraph
  from dask.optimization import cull, fuse
  from dask.utils import ensure_dict
label_desc: Use image node_id 1 for calling a global function with example usage: optimize(dsk, keys) and returns: dsk, dsk