code: def initialize_write(
      cls,
      df,
      fs,
      path,
      append=False,
      partition_on=None,
      ignore_divisions=False,
      division_info=None,
      schema="infer",
      index_cols=None,
      **kwargs,
  ):
      if schema == "infer" or isinstance(schema, dict):
          # Start with schema from _meta_nonempty
          inferred_schema = pyarrow_schema_dispatch(
              df._meta_nonempty.set_index(index_cols)
              if index_cols
              else df._meta_nonempty
          ).remove_metadata()
  
          # Use dict to update our inferred schema
          if isinstance(schema, dict):
              schema = pa.schema(schema)
              for name in schema.names:
                  i = inferred_schema.get_field_index(name)
                  j = schema.get_field_index(name)
                  inferred_schema = inferred_schema.set(
                      i, schema.field(j)
                  )
          schema = inferred_schema
  
      # Check that target directory exists
      fs.mkdirs(path, exist_ok=True)
      if append and division_info is None:
          ignore_divisions = True
  
      full_metadata = (
          None  # metadata for the full dataset, from _metadata
      )
      tail_metadata = (
          None  # metadata for at least the last file in the dataset
      )
      i_offset = 0
      metadata_file_exists = False
      if append:
          # Extract metadata and get file offset if appending
          ds = pa_ds.dataset(
              path, filesystem=_wrapped_fs(fs), format="parquet"
          )
          i_offset = len(ds.files)
          if i_offset > 0:
              try:
                  with fs.open(
                      fs.sep.join([path, "_metadata"]), mode="rb"
                  ) as fil:
                      full_metadata = pq.read_metadata(fil)
                  tail_metadata = full_metadata
                  metadata_file_exists = True
              except OSError:
                  try:
                      with fs.open(
                          sorted(ds.files, key=natural_sort_key)[-1],
                          mode="rb",
                      ) as fil:
                          tail_metadata = pq.read_metadata(fil)
                  except OSError:
                      pass
          else:
              append = (
                  False  # No existing files, can skip the append logic
              )
  
      # If appending, validate against the initial metadata file (if present)
      if append and tail_metadata is not None:
          arrow_schema = tail_metadata.schema.to_arrow_schema()
          names = arrow_schema.names
          has_pandas_metadata = (
              arrow_schema.metadata is not None
              and b"pandas" in arrow_schema.metadata
          )
          if has_pandas_metadata:
              pandas_metadata = json.loads(
                  arrow_schema.metadata[b"pandas"].decode("utf8")
              )
              categories = [
                  c["name"]
                  for c in pandas_metadata["columns"]
                  if c["pandas_type"] == "categorical"
              ]
          else:
              categories = None
          dtypes = _get_pyarrow_dtypes(arrow_schema, categories)
          if set(names) != set(df.columns) - set(partition_on):
              raise ValueError(
                  "Appended columns not the same.\n"
                  "Previous: {} | New: {}".format(
                      names, list(df.columns)
                  )
              )
          elif (pd.Series(dtypes).loc[names] != df[names].dtypes).any():
              # TODO Coerce values for compatible but different dtypes
              raise ValueError(
                  "Appended dtypes differ.\n{}".format(
                      set(dtypes.items()) ^ set(df.dtypes.items())
                  )
              )
  
          # Check divisions if necessary
          if division_info["name"] not in names:
              ignore_divisions = True
          if not ignore_divisions:
              old_end = None
              row_groups = (
                  tail_metadata.row_group(i)
                  for i in range(tail_metadata.num_row_groups)
              )
              index_col_i = names.index(division_info["name"])
              for row_group in row_groups:
                  column = row_group.column(index_col_i)
                  if column.statistics:
                      if old_end is None:
                          old_end = column.statistics.max
                      elif column.statistics.max > old_end:
                          old_end = column.statistics.max
                      else:
                          # Existing column on disk isn't sorted, set
                          # `old_end = None` to skip check below
                          old_end = None
                          break
  
              divisions = division_info["divisions"]
              if old_end is not None and divisions[0] <= old_end:
                  raise ValueError(
                      "The divisions of the appended dataframe overlap with "
                      "previously written divisions. If this is desired, set "
                      "``ignore_divisions=True`` to append anyway.\n"
                      "- End of last written partition: {old_end}\n"
                      "- Start of first new partition: {divisions[0]}"
                  )
  
      extra_write_kwargs = {"schema": schema, "index_cols": index_cols}
      return (
          i_offset,
          full_metadata,
          metadata_file_exists,
          extra_write_kwargs,
      )
  
imports: from __future__ import annotations
  import json
  import operator
  import textwrap
  import warnings
  from collections import defaultdict
  from datetime import datetime
  from functools import reduce
  import numpy
  import pandas
  import pyarrow
  import pyarrow.parquet
  from fsspec.core import expand_paths_if_needed, stringify_path
  from fsspec.implementations.arrow import ArrowFSWrapper
  from pyarrow import dataset
  from pyarrow import fs
  import dask
  from dask.base import tokenize
  from dask.core import flatten
  from dask.dataframe.backends import pyarrow_schema_dispatch
  from dask.dataframe.io.parquet.utils import Engine, _get_aggregation_depth, _infer_split_row_groups, _normalize_index_columns, _process_open_file_options, _row_groups_to_parts, _set_gather_statistics, _set_metadata_task_size, _sort_and_analyze_paths
  from dask.dataframe.io.utils import _get_pyarrow_dtypes, _is_local_fs, _open_input_files
  from dask.dataframe.utils import clear_known_categories, pyarrow_strings_enabled
  from dask.delayed import Delayed
  from dask.utils import getargspec, natural_sort_key
label_desc: Use image node_id 5 for calling the ArrowDatasetEngine obj's underlying member method code with example usage: obj.initialize_write(cls, df, fs, path, append, partition_on, ignore_divisions, division_info, schema, index_cols) and returns: i_offset,  full_metadata,  metadata_file_exists,  extra_write_kwargs