code: def to_hdf(
      df,
      path,
      key,
      mode="a",
      append=False,
      scheduler=None,
      name_function=None,
      compute=True,
      lock=None,
      dask_kwargs=None,
      **kwargs,
  ):
      """Store Dask Dataframe to Hierarchical Data Format (HDF) files
  
      This is a parallel version of the Pandas function of the same name.  Please
      see the Pandas docstring for more detailed information about shared keyword
      arguments.
  
      This function differs from the Pandas version by saving the many partitions
      of a Dask DataFrame in parallel, either to many files, or to many datasets
      within the same file.  You may specify this parallelism with an asterix
      ``*`` within the filename or datapath, and an optional ``name_function``.
      The asterix will be replaced with an increasing sequence of integers
      starting from ``0`` or with the result of calling ``name_function`` on each
      of those integers.
  
      This function only supports the Pandas ``'table'`` format, not the more
      specialized ``'fixed'`` format.
  
      Parameters
      ----------
      path : string, pathlib.Path
          Path to a target filename. Supports strings, ``pathlib.Path``, or any
          object implementing the ``__fspath__`` protocol. May contain a ``*`` to
          denote many filenames.
      key : string
          Datapath within the files.  May contain a ``*`` to denote many locations
      name_function : function
          A function to convert the ``*`` in the above options to a string.
          Should take in a number from 0 to the number of partitions and return a
          string. (see examples below)
      compute : bool
          Whether or not to execute immediately.  If False then this returns a
          ``dask.Delayed`` value.
      lock : bool, Lock, optional
          Lock to use to prevent concurrency issues.  By default a
          ``threading.Lock``, ``multiprocessing.Lock`` or ``SerializableLock``
          will be used depending on your scheduler if a lock is required. See
          dask.utils.get_scheduler_lock for more information about lock
          selection.
      scheduler : string
          The scheduler to use, like "threads" or "processes"
      **other:
          See pandas.to_hdf for more information
  
      Examples
      --------
      Save Data to a single file
  
      >>> df.to_hdf('output.hdf', '/data')            # doctest: +SKIP
  
      Save data to multiple datapaths within the same file:
  
      >>> df.to_hdf('output.hdf', '/data-*')          # doctest: +SKIP
  
      Save data to multiple files:
  
      >>> df.to_hdf('output-*.hdf', '/data')          # doctest: +SKIP
  
      Save data to multiple files, using the multiprocessing scheduler:
  
      >>> df.to_hdf('output-*.hdf', '/data', scheduler='processes') # doctest: +SKIP
  
      Specify custom naming scheme.  This writes files as
      '2000-01-01.hdf', '2000-01-02.hdf', '2000-01-03.hdf', etc..
  
      >>> from datetime import date, timedelta
      >>> base = date(year=2000, month=1, day=1)
      >>> def name_function(i):
      ...     ''' Convert integer 0 to n to a string '''
      ...     return base + timedelta(days=i)
  
      >>> df.to_hdf('*.hdf', '/data', name_function=name_function) # doctest: +SKIP
  
      Returns
      -------
      filenames : list
          Returned if ``compute`` is True. List of file names that each partition
          is saved to.
      delayed : dask.Delayed
          Returned if ``compute`` is False. Delayed object to execute ``to_hdf``
          when computed.
  
      See Also
      --------
      read_hdf:
      to_parquet:
      """
      if dask_kwargs is None:
          dask_kwargs = {}
  
      name = "to-hdf-" + uuid.uuid1().hex
  
      pd_to_hdf = df._partition_type.to_hdf
  
      single_file = True
      single_node = True
  
      path = stringify_path(path)
  
      # if path is string, format using i_name
      if isinstance(path, str):
          if path.count("*") + key.count("*") > 1:
              raise ValueError(
                  "A maximum of one asterisk is accepted in file path and dataset key"
              )
  
          fmt_obj = lambda path, i_name: path.replace("*", i_name)
  
          if "*" in path:
              single_file = False
      else:
          if key.count("*") > 1:
              raise ValueError(
                  "A maximum of one asterisk is accepted in dataset key"
              )
  
          fmt_obj = lambda path, _: path
  
      if "*" in key:
          single_node = False
  
      if "format" in kwargs and kwargs["format"] not in ["t", "table"]:
          raise ValueError(
              "Dask only support 'table' format in hdf files."
          )
  
      if mode not in ("a", "w", "r+"):
          raise ValueError("Mode must be one of 'a', 'w' or 'r+'")
  
      if name_function is None:
          name_function = build_name_function(df.npartitions - 1)
  
      # we guarantee partition order is preserved when its saved and read
      # so we enforce name_function to maintain the order of its input.
      if not (single_file and single_node):
          formatted_names = [
              name_function(i) for i in range(df.npartitions)
          ]
          if formatted_names != sorted(formatted_names):
              warn(
                  "To preserve order between partitions name_function "
                  "must preserve the order of its input"
              )
  
      # If user did not specify scheduler and write is sequential default to the
      # sequential scheduler. otherwise let the _get method choose the scheduler
      try:
          from distributed import default_client
  
          default_client()
          client_available = True
      except (ImportError, ValueError):
          client_available = False
  
      if (
          scheduler is None
          and not config.get("scheduler", None)
          and not client_available
          and single_node
          and single_file
      ):
          scheduler = "single-threaded"
  
      # handle lock default based on whether we're writing to a single entity
      _actual_get = get_scheduler(collections=[df], scheduler=scheduler)
      if lock is None:
          if not single_node:
              lock = True
          elif not single_file and _actual_get is not MP_GET:
              # if we're writing to multiple files with the multiprocessing
              # scheduler we don't need to lock
              lock = True
          else:
              lock = False
  
      # TODO: validation logic to ensure that provided locks are compatible with the scheduler
      if isinstance(lock, bool) and lock:
          lock = get_scheduler_lock(df, scheduler=scheduler)
      elif lock:
          assert isinstance(lock, SupportsLock)
      dsk = dict()
  
      i_name = name_function(0)
      kwargs.update(
          {
              "format": "table",
              "mode": mode,
              "append": append,
              "key": key.replace("*", i_name),
          }
      )
      dsk[(name, 0)] = (
          _pd_to_hdf,
          pd_to_hdf,
          lock,
          [(df._name, 0), fmt_obj(path, i_name)],
          kwargs,
      )
  
      kwargs2 = kwargs.copy()
      if single_file:
          kwargs2["mode"] = "a"
      if single_node:
          kwargs2["append"] = True
  
      filenames = []
      for i in range(0, df.npartitions):
          i_name = name_function(i)
          filenames.append(fmt_obj(path, i_name))
  
      for i in range(1, df.npartitions):
          i_name = name_function(i)
          kwargs2["key"] = key.replace("*", i_name)
          task = (
              _pd_to_hdf,
              pd_to_hdf,
              lock,
              [(df._name, i), fmt_obj(path, i_name)],
              kwargs2.copy(),
          )
          if single_file:
              link_dep = i - 1 if single_node else 0
              task = (_link, (name, link_dep), task)
          dsk[(name, i)] = task
  
      if single_file and single_node:
          keys = [(name, df.npartitions - 1)]
      else:
          keys = [(name, i) for i in range(df.npartitions)]
  
      final_name = name + "-final"
      dsk[(final_name, 0)] = (lambda x: None, keys)
      graph = HighLevelGraph.from_collections(
          (name, 0), dsk, dependencies=[df]
      )
  
      if compute:
          compute_as_if_collection(
              DataFrame, graph, keys, scheduler=scheduler, **dask_kwargs
          )
          return filenames
      else:
          return Scalar(graph, final_name, "")
  
imports: from __future__ import annotations
  import os
  import uuid
  from fnmatch import fnmatch
  from glob import glob
  from warnings import warn
  import pandas
  from fsspec.utils import build_name_function, stringify_path
  from tlz import merge
  from dask import config
  from dask.base import compute_as_if_collection, get_scheduler, named_schedulers, tokenize
  from dask.dataframe.backends import dataframe_creation_dispatch
  from dask.dataframe.core import DataFrame, Scalar
  from dask.dataframe.io.io import _link, from_map
  from dask.dataframe.io.utils import DataFrameIOFunction, SupportsLock
  from dask.highlevelgraph import HighLevelGraph
  from dask.utils import get_scheduler_lock
  from dask.dataframe.core import _Frame
label_desc: Use image node_id 2 for calling a global function with example usage: to_hdf(df, path, key, mode, append, scheduler, name_function, compute, lock, dask_kwargs) and returns: filenames, Scalar