class_docstr: Write a parquet file or dataset incrementally
  
  Parameters
  ----------
  path : str
      A local directory path or S3 URL. Will be used as root directory
      path while writing a partitioned dataset.
  partition_cols : list
      Column names by which to partition the dataset
      Columns are partitioned in the order they are given
  index : bool, default None
      If ``True``, include the dataframe's index(es) in the file output.
      If ``False``, they will not be written to the file. If ``None``,
      index(es) other than RangeIndex will be saved as columns.
  compression : {'snappy', None}, default 'snappy'
      Name of the compression to use. Use ``None`` for no compression.
  statistics : {'ROWGROUP', 'PAGE', 'COLUMN', 'NONE'}, default 'ROWGROUP'
      Level at which column statistics should be included in file.
  max_file_size : int or str, default None
      A file size that cannot be exceeded by the writer.
      It is in bytes, if the input is int.
      Size can also be a str in form or "10 MB", "1 GB", etc.
      If this parameter is used, it is mandatory to pass
      `file_name_prefix`.
  file_name_prefix : str
      This is a prefix to file names generated only when
      `max_file_size` is specified.
  storage_options : dict, optional, default None
      Extra options that make sense for a particular storage connection,
      e.g. host, port, username, password, etc. For HTTP(S) URLs the
      key-value pairs are forwarded to ``urllib.request.Request`` as
      header options. For other URLs (e.g. starting with "s3://", and
      "gcs://") the key-value pairs are forwarded to ``fsspec.open``.
      Please see ``fsspec`` and ``urllib`` for more details.
  
  
  Examples
  --------
  Using a context
  
  >>> df1 = cudf.DataFrame({"a": [1, 1, 2, 2, 1], "b": [9, 8, 7, 6, 5]})
  >>> df2 = cudf.DataFrame({"a": [1, 3, 3, 1, 3], "b": [4, 3, 2, 1, 0]})
  >>> with ParquetDatasetWriter("./dataset", partition_cols=["a"]) as cw:
  ...     cw.write_table(df1)
  ...     cw.write_table(df2)
  
  By manually calling ``close()``
  
  >>> cw = ParquetDatasetWriter("./dataset", partition_cols=["a"])
  >>> cw.write_table(df1)
  >>> cw.write_table(df2)
  >>> cw.close()
  
  Both the methods will generate the same directory structure
  
  .. code-block:: none
  
      dataset/
          a=1
              <filename>.parquet
          a=2
              <filename>.parquet
          a=3
              <filename>.parquet
class_docstr_tok: ["Write","a","parquet","file","or","dataset","incrementally","Parameters","--","--","--","--","--","path",":","str","A","local","directory","path","or","S3","URL",".","Will","be","used","as","root","directory","path","while","writing","a","partitioned","dataset",".","partition_cols",":","list","Column","names","by","which","to","partition","the","dataset","Columns","are","partitioned","in","the","order","they","are","given","index",":","bool",",","default","None","If","``","True","``",",","include","the","dataframe","'s","index","(","es",")","in","the","file","output",".","If","``","False","``",",","they","will","not","be","written","to","the","file",".","If","``","None","``",",","index","(","es",")","other","than","RangeIndex","will","be","saved","as","columns",".","compression",":","{","'snappy","'",",","None","}",",","default","'snappy'","Name","of","the","compression","to","use",".","Use","``","None","``","for","no","compression",".","statistics",":","{","'ROWGROUP","'",",","'PAGE","'",",","'COLUMN","'",",","'NONE","'","}",",","default","'ROWGROUP'","Level","at","which","column","statistics","should","be","included","in","file",".","max_file_size",":","int","or","str",",","default","None","A","file","size","that","can","not","be","exceeded","by","the","writer",".","It","is","in","bytes",",","if","the","input","is","int",".","Size","can","also","be","a","str","in","form","or","``","10","MB","''",",","``","1","GB","''",",","etc",".","If","this","parameter","is","used",",","it","is","mandatory","to","pass","`","file_name_prefix","`",".","file_name_prefix",":","str","This","is","a","prefix","to","file","names","generated","only","when","`","max_file_size","`","is","specified",".","storage_options",":","dict",",","optional",",","default","None","Extra","options","that","make","sense","for","a","particular","storage","connection",",","e.g",".","host",",","port",",","username",",","password",",","etc",".","For","HTTP","(","S",")","URLs","the","key-value","pairs","are","forwarded","to","``","urllib.request.Request","``","as","header","options",".","For","other","URLs","(","e.g",".","starting","with","``","s3",":","\/\/","''",",","and","``","gcs",":","\/\/","''",")","the","key-value","pairs","are","forwarded","to","``","fsspec.open","``",".","Please","see","``","fsspec","``","and","``","urllib","``","for","more","details",".","Examples","--","--","--","--","Using","a","context",">",">",">","df1","=","cudf.DataFrame","(","{","``","a","''",":","[","1",",","1",",","2",",","2",",","1","]",",","``","b","''",":","[","9",",","8",",","7",",","6",",","5","]","}",")",">",">",">","df2","=","cudf.DataFrame","(","{","``","a","''",":","[","1",",","3",",","3",",","1",",","3","]",",","``","b","''",":","[","4",",","3",",","2",",","1",",","0","]","}",")",">",">",">","with","ParquetDatasetWriter","(","``",".\/dataset","''",",","partition_cols=","[","``","a","''","]",")","as","cw",":","...","cw.write_table","(","df1",")","...","cw.write_table","(","df2",")","By","manually","calling","``","close","(",")","``",">",">",">","cw","=","ParquetDatasetWriter","(","``",".\/dataset","''",",","partition_cols=","[","``","a","''","]",")",">",">",">","cw.write_table","(","df1",")",">",">",">","cw.write_table","(","df2",")",">",">",">","cw.close","(",")","Both","the","methods","will","generate","the","same","directory","structure","..","code-block",":",":","none","dataset\/","a=1","<","filename",">",".parquet","a=2","<","filename",">",".parquet","a=3","<","filename",">",".parquet"]
docstr: Write a dataframe to the file/dataset
docstr_tok: ["Write","a","dataframe","to","the","file\/dataset"]
code: def write_table(self, df):
      """
      Write a dataframe to the file/dataset
      """
      (
          part_names,
          grouped_df,
          part_offsets,
      ) = _get_groups_and_offsets(
          df=df,
          partition_cols=self.partition_cols,
          preserve_index=self.common_args["index"],
      )
      fs = ioutils._ensure_filesystem(None, self.path, None)
      fs.mkdirs(self.path, exist_ok=True)
  
      full_paths = []
      metadata_file_paths = []
      full_offsets = [0]
  
      for idx, keys in enumerate(part_names.itertuples(index=False)):
          subdir = fs.sep.join(
              [
                  f"{name}={val}"
                  for name, val in zip(self.partition_cols, keys)
              ]
          )
          prefix = fs.sep.join([self.path, subdir])
          fs.mkdirs(prefix, exist_ok=True)
          current_offset = (part_offsets[idx], part_offsets[idx + 1])
          num_chunks = 1
          parts = 1
  
          if self.max_file_size is not None:
              # get the current partition
              start, end = current_offset
              sliced_df = grouped_df[start:end]
  
              current_file_size = _get_estimated_file_size(sliced_df)
              if current_file_size > self.max_file_size:
                  # if the file is too large, compute metadata for
                  # smaller chunks
                  parts = math.ceil(
                      current_file_size / self.max_file_size
                  )
                  new_offsets = list(
                      range(start, end, int((end - start) / parts))
                  )[1:]
                  new_offsets.append(end)
                  num_chunks = len(new_offsets)
                  parts = len(new_offsets)
                  full_offsets.extend(new_offsets)
              else:
                  full_offsets.append(end)
  
              curr_file_num = 0
              num_chunks = 0
              while num_chunks < parts:
                  new_file_name = (
                      f"{self.filename}_{curr_file_num}.parquet"
                  )
                  new_full_path = fs.sep.join([prefix, new_file_name])
  
                  # Check if the same `new_file_name` exists and
                  # generate a `new_file_name`
                  while new_full_path in self._file_sizes and (
                      self._file_sizes[new_full_path]
                      + (current_file_size / parts)
                  ) > (self.max_file_size):
                      curr_file_num += 1
                      new_file_name = (
                          f"{self.filename}_{curr_file_num}.parquet"
                      )
                      new_full_path = fs.sep.join(
                          [prefix, new_file_name]
                      )
  
                  self._file_sizes[
                      new_full_path
                  ] = self._file_sizes.get(new_full_path, 0) + (
                      current_file_size / parts
                  )
                  full_paths.append(new_full_path)
                  metadata_file_paths.append(
                      fs.sep.join([subdir, new_file_name])
                  )
                  num_chunks += 1
                  curr_file_num += 1
          else:
              self.filename = self.filename or _generate_filename()
              full_path = fs.sep.join([prefix, self.filename])
              full_paths.append(full_path)
              metadata_file_paths.append(
                  fs.sep.join([subdir, self.filename])
              )
              full_offsets.append(current_offset[1])
  
      paths, metadata_file_paths, offsets = (
          full_paths,
          metadata_file_paths,
          full_offsets,
      )
      existing_cw_batch = defaultdict(dict)
      new_cw_paths = []
      partition_info = [
          (i, j - i) for i, j in zip(offsets, offsets[1:])
      ]
  
      for path, part_info, meta_path in zip(
          paths,
          partition_info,
          metadata_file_paths,
      ):
          if path in self.path_cw_map:  # path is a currently open file
              cw_idx = self.path_cw_map[path]
              existing_cw_batch[cw_idx][path] = part_info
          else:  # path not currently handled by any chunked writer
              new_cw_paths.append((path, part_info, meta_path))
  
      # Write out the parts of grouped_df currently handled by existing cw's
      for cw_idx, path_to_part_info_map in existing_cw_batch.items():
          cw = self._chunked_writers[cw_idx][0]
          # match found paths with this cw's paths and nullify partition info
          # for partition_col values not in this batch
          this_cw_part_info = [
              path_to_part_info_map.get(path, (0, 0))
              for path in self._chunked_writers[cw_idx][1]
          ]
          cw.write_table(grouped_df, this_cw_part_info)
  
      if new_cw_paths:
          # Create new cw for unhandled paths encountered in this write_table
          new_paths, part_info, meta_paths = zip(*new_cw_paths)
          self._chunked_writers.append(
              (
                  ParquetWriter(new_paths, **self.common_args),
                  new_paths,
                  meta_paths,
              )
          )
          new_cw_idx = len(self._chunked_writers) - 1
          self.path_cw_map.update({k: new_cw_idx for k in new_paths})
          self._chunked_writers[-1][0].write_table(
              grouped_df, part_info
          )
  
imports: from __future__ import annotations
  import itertools
  import math
  import operator
  import shutil
  import tempfile
  import warnings
  from collections import defaultdict
  from contextlib import ExitStack
  from functools import partial, reduce
  from typing import Callable, Dict, List, Optional, Tuple
  from uuid import uuid4
  import numpy
  import pandas
  from pyarrow import dataset
  import cudf
  from cudf._lib import parquet
  from cudf.api.types import is_list_like
  from cudf.core.column import build_categorical_column, column_empty, full
  from cudf.utils import ioutils
  from cudf.utils.nvtx_annotation import _cudf_nvtx_annotate
label_desc: Use image node_id 2 for calling the ParquetDatasetWriter obj's underlying member method code with example usage: obj.write_table(df) without return types