code: def to_parquet(
      df,
      path,
      engine="auto",
      compression="snappy",
      write_index=True,
      append=False,
      overwrite=False,
      ignore_divisions=False,
      partition_on=None,
      storage_options=None,
      custom_metadata=None,
      write_metadata_file=None,
      compute=True,
      compute_kwargs=None,
      schema="infer",
      name_function=None,
      filesystem=None,
      **kwargs,
  ):
      """Store Dask.dataframe to Parquet files
  
      Notes
      -----
      Each partition will be written to a separate file.
  
      Parameters
      ----------
      df : dask.dataframe.DataFrame
      path : string or pathlib.Path
          Destination directory for data.  Prepend with protocol like ``s3://``
          or ``hdfs://`` for remote data.
      engine : {'auto', 'pyarrow'}
          Parquet library to use. Defaults to 'auto', which uses ``pyarrow`` if
          it is installed, and falls back to the deprecated ``fastparquet`` otherwise.
          Note that ``fastparquet`` does not support all functionality offered by
          ``pyarrow``.
          This is also used by third-party packages (e.g. CuDF) to inject bespoke engines.
      compression : string or dict, default 'snappy'
          Either a string like ``"snappy"`` or a dictionary mapping column names
          to compressors like ``{"name": "gzip", "values": "snappy"}``. Defaults
          to ``"snappy"``.
      write_index : boolean, default True
          Whether or not to write the index. Defaults to True.
      append : bool, default False
          If False (default), construct data-set from scratch. If True, add new
          row-group(s) to an existing data-set. In the latter case, the data-set
          must exist, and the schema must match the input data.
      overwrite : bool, default False
          Whether or not to remove the contents of `path` before writing the dataset.
          The default is False.  If True, the specified path must correspond to
          a directory (but not the current working directory).  This option cannot
          be set to True if `append=True`.
          NOTE: `overwrite=True` will remove the original data even if the current
          write operation fails.  Use at your own risk.
      ignore_divisions : bool, default False
          If False (default) raises error when previous divisions overlap with
          the new appended divisions. Ignored if append=False.
      partition_on : list, default None
          Construct directory-based partitioning by splitting on these fields'
          values. Each dask partition will result in one or more datafiles,
          there will be no global groupby.
      storage_options : dict, default None
          Key/value pairs to be passed on to the file-system backend, if any.
      custom_metadata : dict, default None
          Custom key/value metadata to include in all footer metadata (and
          in the global "_metadata" file, if applicable).  Note that the custom
          metadata may not contain the reserved b"pandas" key.
      write_metadata_file : bool or None, default None
          Whether to write the special ``_metadata`` file. If ``None`` (the
          default), a ``_metadata`` file will only be written if ``append=True``
          and the dataset already has a ``_metadata`` file.
      compute : bool, default True
          If ``True`` (default) then the result is computed immediately. If
          ``False`` then a ``dask.dataframe.Scalar`` object is returned for
          future computation.
      compute_kwargs : dict, default True
          Options to be passed in to the compute method
      schema : pyarrow.Schema, dict, "infer", or None, default "infer"
          Global schema to use for the output dataset. Defaults to "infer", which
          will infer the schema from the dask dataframe metadata. This is usually
          sufficient for common schemas, but notably will fail for ``object``
          dtype columns that contain things other than strings. These columns
          will require an explicit schema be specified. The schema for a subset
          of columns can be overridden by passing in a dict of column names to
          pyarrow types (for example ``schema={"field": pa.string()}``); columns
          not present in this dict will still be automatically inferred.
          Alternatively, a full ``pyarrow.Schema`` may be passed, in which case
          no schema inference will be done. Passing in ``schema=None`` will
          disable the use of a global file schema - each written file may use a
          different schema dependent on the dtypes of the corresponding
          partition.
      name_function : callable, default None
          Function to generate the filename for each output partition.
          The function should accept an integer (partition index) as input and
          return a string which will be used as the filename for the corresponding
          partition. Should preserve the lexicographic order of partitions.
          If not specified, files will created using the convention
          ``part.0.parquet``, ``part.1.parquet``, ``part.2.parquet``, ...
          and so on for each partition in the DataFrame.
      filesystem: "fsspec", "arrow", or fsspec.AbstractFileSystem backend to use.
      **kwargs :
          Extra options to be passed on to the specific backend.
  
      Examples
      --------
      >>> df = dd.read_csv(...)  # doctest: +SKIP
      >>> df.to_parquet('/path/to/output/', ...)  # doctest: +SKIP
  
      By default, files will be created in the specified output directory using the
      convention ``part.0.parquet``, ``part.1.parquet``, ``part.2.parquet``, ... and so on for
      each partition in the DataFrame. To customize the names of each file, you can use the
      ``name_function=`` keyword argument. The function passed to ``name_function`` will be
      used to generate the filename for each partition and should expect a partition's index
      integer as input and return a string which will be used as the filename for the corresponding
      partition. Strings produced by ``name_function`` must preserve the order of their respective
      partition indices.
  
      For example:
  
      >>> name_function = lambda x: f"data-{x}.parquet"
      >>> df.to_parquet('/path/to/output/', name_function=name_function)  # doctest: +SKIP
  
      will result in the following files being created::
  
          /path/to/output/
              ├── data-0.parquet
              ├── data-1.parquet
              ├── data-2.parquet
              └── ...
  
      See Also
      --------
      read_parquet: Read parquet data to dask.dataframe
      """
      compute_kwargs = compute_kwargs or {}
  
      partition_on = partition_on or []
      if isinstance(partition_on, str):
          partition_on = [partition_on]
  
      if set(partition_on) - set(df.columns):
          raise ValueError(
              "Partitioning on non-existent column. "
              "partition_on=%s ."
              "columns=%s" % (str(partition_on), str(list(df.columns)))
          )
  
      if df.columns.inferred_type not in {"string", "empty"}:
          raise ValueError(
              "parquet doesn't support non-string column names"
          )
  
      engine = get_engine(engine)
  
      if hasattr(path, "name"):
          path = stringify_path(path)
  
      fs, _paths, _, _ = engine.extract_filesystem(
          path,
          filesystem=filesystem,
          dataset_options={},
          open_file_options={},
          storage_options=storage_options,
      )
      assert len(_paths) == 1, "only one path"
      path = _paths[0]
  
      if overwrite:
          if append:
              raise ValueError(
                  "Cannot use both `overwrite=True` and `append=True`!"
              )
  
          if fs.exists(path) and fs.isdir(path):
              # Check for any previous parquet layers reading from a file in the
              # output directory, since deleting those files now would result in
              # errors or incorrect results.
              for layer_name, layer in df.dask.layers.items():
                  if layer_name.startswith(
                      "read-parquet-"
                  ) and isinstance(layer, DataFrameIOLayer):
                      path_with_slash = (
                          path.rstrip("/") + "/"
                      )  # ensure trailing slash
                      for input in layer.inputs:
                          # Note that `input` may be either `dict` or `List[dict]`
                          for piece_dict in (
                              input
                              if isinstance(input, list)
                              else [input]
                          ):
                              if piece_dict["piece"][0].startswith(
                                  path_with_slash
                              ):
                                  raise ValueError(
                                      "Reading and writing to the same parquet file within "
                                      "the same task graph is not supported."
                                  )
  
              # Don't remove the directory if it's the current working directory
              if _is_local_fs(fs):
                  working_dir = fs.expand_path(".")[0]
                  if path.rstrip("/") == working_dir.rstrip("/"):
                      raise ValueError(
                          "Cannot clear the contents of the current working directory!"
                      )
  
              # It's safe to clear the output directory
              fs.rm(path, recursive=True)
  
      # Always skip divisions checks if divisions are unknown
      if not df.known_divisions:
          ignore_divisions = True
  
      # Save divisions and corresponding index name. This is necessary,
      # because we may be resetting the index to write the file
      division_info = {"divisions": df.divisions, "name": df.index.name}
      if division_info["name"] is None:
          # As of 0.24.2, pandas will rename an index with name=None
          # when df.reset_index() is called.  The default name is "index",
          # but dask will always change the name to the NONE_LABEL constant
          if NONE_LABEL not in df.columns:
              division_info["name"] = NONE_LABEL
          elif write_index:
              raise ValueError(
                  "Index must have a name if __null_dask_index__ is a column."
              )
          else:
              warnings.warn(
                  "If read back by Dask, column named __null_dask_index__ "
                  "will be set to the index (and renamed to None)."
              )
  
      # There are some "reserved" names that may be used as the default column
      # name after resetting the index. However, we don't want to treat it as
      # a "special" name if the string is already used as a "real" column name.
      reserved_names = []
      for name in ["index", "level_0"]:
          if name not in df.columns:
              reserved_names.append(name)
  
      # If write_index==True (default), reset the index and record the
      # name of the original index in `index_cols` (we will set the name
      # to the NONE_LABEL constant if it is originally `None`).
      # `fastparquet` will use `index_cols` to specify the index column(s)
      # in the metadata.  `pyarrow` will revert the `reset_index` call
      # below if `index_cols` is populated (because pyarrow will want to handle
      # index preservation itself).  For both engines, the column index
      # will be written to "pandas metadata" if write_index=True
      index_cols = []
      if write_index:
          real_cols = set(df.columns)
          none_index = list(df._meta.index.names) == [None]
          df = df.reset_index()
          if none_index:
              df.columns = [
                  c if c not in reserved_names else NONE_LABEL
                  for c in df.columns
              ]
          index_cols = [c for c in set(df.columns) - real_cols]
      else:
          # Not writing index - might as well drop it
          df = df.reset_index(drop=True)
  
      if custom_metadata and b"pandas" in custom_metadata.keys():
          raise ValueError(
              "User-defined key/value metadata (custom_metadata) can not "
              "contain a b'pandas' key.  This key is reserved by Pandas, "
              "and overwriting the corresponding value can render the "
              "entire dataset unreadable."
          )
  
      # Engine-specific initialization steps to write the dataset.
      # Possibly create parquet metadata, and load existing stuff if appending
      (
          i_offset,
          fmd,
          metadata_file_exists,
          extra_write_kwargs,
      ) = engine.initialize_write(
          df,
          fs,
          path,
          append=append,
          ignore_divisions=ignore_divisions,
          partition_on=partition_on,
          division_info=division_info,
          index_cols=index_cols,
          schema=schema,
          custom_metadata=custom_metadata,
          **kwargs,
      )
  
      # By default we only write a metadata file when appending if one already
      # exists
      if append and write_metadata_file is None:
          write_metadata_file = metadata_file_exists
  
      # Check that custom name_function is valid,
      # and that it will produce unique names
      if name_function is not None:
          if not callable(name_function):
              raise ValueError(
                  "``name_function`` must be a callable with one argument."
              )
          filenames = [
              name_function(i + i_offset) for i in range(df.npartitions)
          ]
          if len(set(filenames)) < len(filenames):
              raise ValueError(
                  "``name_function`` must produce unique filenames."
              )
  
      # If we are using a remote filesystem and retries is not set, bump it
      # to be more fault tolerant, as transient transport errors can occur.
      # The specific number 5 isn't hugely motivated: it's less than ten and more
      # than two.
      annotations = dask.get_annotations()
      if "retries" not in annotations and not _is_local_fs(fs):
          ctx = dask.annotate(retries=5)
      else:
          ctx = contextlib.nullcontext()
  
      # Create Blockwise layer for parquet-data write
      with ctx:
          data_write = df.map_partitions(
              ToParquetFunctionWrapper(
                  engine,
                  path,
                  fs,
                  partition_on,
                  write_metadata_file,
                  i_offset,
                  name_function,
                  toolz.merge(
                      kwargs,
                      {
                          "compression": compression,
                          "custom_metadata": custom_metadata,
                      },
                      extra_write_kwargs,
                  ),
              ),
              BlockIndex((df.npartitions,)),
              # Pass in the original metadata to avoid
              # metadata emulation in `map_partitions`.
              # This is necessary, because we are not
              # expecting a dataframe-like output.
              meta=df._meta,
              enforce_metadata=False,
              transform_divisions=False,
              align_dataframes=False,
          )
  
      # Collect metadata and write _metadata.
      # TODO: Use tree-reduction layer (when available)
      if write_metadata_file:
          final_name = "metadata-" + data_write._name
          dsk = {
              (final_name, 0): (
                  apply,
                  engine.write_metadata,
                  [
                      data_write.__dask_keys__(),
                      fmd,
                      fs,
                      path,
                  ],
                  {"append": append, "compression": compression},
              )
          }
      else:
          # NOTE: We still define a single task to tie everything together
          # when we are not writing a _metadata file. We do not want to
          # return `data_write` (or a `data_write.to_bag()`), because calling
          # `compute()` on a multi-partition collection requires the overhead
          # of trying to concatenate results on the client.
          final_name = "store-" + data_write._name
          dsk = {
              (final_name, 0): (
                  lambda x: None,
                  data_write.__dask_keys__(),
              )
          }
  
      # Convert data_write + dsk to computable collection
      graph = HighLevelGraph.from_collections(
          final_name, dsk, dependencies=(data_write,)
      )
      out = Scalar(graph, final_name, "")
  
      if compute:
          out = out.compute(**compute_kwargs)
  
      # Invalidate the filesystem listing cache for the output path after write.
      # We do this before returning, even if `compute=False`. This helps ensure
      # that reading files that were just written succeeds.
      fs.invalidate_cache(path)
  
      return out
  
imports: from __future__ import annotations
  import contextlib
  import math
  import warnings
  from typing import Literal
  import pandas
  import tlz
  from fsspec.core import get_fs_token_paths
  from fsspec.utils import stringify_path
  import dask
  from dask.base import tokenize
  from dask.blockwise import BlockIndex
  from dask.dataframe.backends import dataframe_creation_dispatch
  from dask.dataframe.core import DataFrame, Scalar
  from dask.dataframe.io.io import from_map
  from dask.dataframe.io.parquet.utils import Engine, _sort_and_analyze_paths, _split_user_options
  from dask.dataframe.io.utils import DataFrameIOFunction, _is_local_fs
  from dask.dataframe.methods import concat
  from dask.delayed import Delayed
  from dask.highlevelgraph import HighLevelGraph
  from dask.layers import DataFrameIOLayer
  from dask.utils import apply, import_required, natural_sort_key, parse_bytes
label_desc: Use image node_id 4 for calling a global function with example usage: to_parquet(df, path, engine, compression, write_index, append, overwrite, ignore_divisions, partition_on, storage_options, custom_metadata, write_metadata_file, compute, compute_kwargs, schema, name_function, filesystem) and returns: out