code: def _write_samples_to_parquet_coo(
      results: cudf.DataFrame,
      offsets: cudf.DataFrame,
      renumber_map: cudf.DataFrame,
      batches_per_partition: int,
      output_path: str,
      partition_info: Optional[Union[dict, str]] = None,
  ) -> cudf.Series:
      """
      Writes COO compressed samples to parquet.
  
      Batches that are empty are discarded, and the remaining non-empty
      batches are renumbered to be contiguous starting from the first
      batch id.  This means that the output batch ids may not match
      the input batch ids.
  
      results: cudf.DataFrame
          The results dataframe containing the sampled minibatches.
      offsets: cudf.DataFrame
          The offsets dataframe indicating the start/end of each minibatch
          in the reuslts dataframe.
      renumber_map: cudf.DataFrame
          The renumber map containing the mapping of renumbered vertex ids
          to original vertex ids.
      batches_per_partition: int
          The maximum number of minibatches allowed per written parquet partition.
      output_path: str
          The output path (where parquet files should be written to).
      partition_info: Union[dict, str]
          Either a dictionary containing partition data from dask, the string 'sg'
          indicating that this is a single GPU write, or None indicating that this
          function should perform a no-op (required by dask).
  
      Returns an empty cudf series.
      """
  
      # Required by dask; need to skip dummy partitions.
      if partition_info is None or len(results) == 0:
          return cudf.Series(dtype="int64")
      if partition_info != "sg" and (
          not isinstance(partition_info, dict)
      ):
          raise ValueError("Invalid value of partition_info")
  
      offsets = offsets[:-1]
  
      # Offsets is always in order, so the last batch id is always the highest
      max_batch_id = offsets.batch_id.iloc[-1]
      results.dropna(axis=1, how="all", inplace=True)
      results["hop_id"] = results["hop_id"].astype("uint8")
  
      for p in range(0, len(offsets), batches_per_partition):
          offsets_p = offsets.iloc[p : p + batches_per_partition]
          start_batch_id = offsets_p.batch_id.iloc[0]
          end_batch_id = offsets_p.batch_id.iloc[len(offsets_p) - 1]
  
          reached_end = end_batch_id == max_batch_id
  
          start_ix = offsets_p.offsets.iloc[0]
          if reached_end:
              end_ix = len(results)
          else:
              offsets_z = offsets[
                  offsets.batch_id == (end_batch_id + 1)
              ]
              end_ix = offsets_z.offsets.iloc[0]
  
          results_p = results.iloc[start_ix:end_ix].reset_index(
              drop=True
          )
  
          if end_batch_id - start_batch_id + 1 > len(offsets_p):
              # This occurs when some batches returned 0 samples.
              # To properly account this, the remaining batches are
              # renumbered to have contiguous batch ids and the empty
              # samples are dropped.
              offsets_p.drop("batch_id", axis=1, inplace=True)
              batch_id_range = cudf.Series(
                  cupy.arange(
                      start_batch_id, start_batch_id + len(offsets_p)
                  )
              )
              end_batch_id = start_batch_id + len(offsets_p) - 1
          else:
              batch_id_range = offsets_p.batch_id
  
          results_p["batch_id"] = batch_id_range.repeat(
              cupy.diff(offsets_p.offsets.values, append=end_ix)
          ).values
  
          if renumber_map is not None:
              renumber_map_start_ix = (
                  offsets_p.renumber_map_offsets.iloc[0]
              )
  
              if reached_end:
                  renumber_map_end_ix = len(renumber_map)
              else:
                  renumber_map_end_ix = (
                      offsets_z.renumber_map_offsets.iloc[0]
                  )
  
              renumber_map_p = renumber_map.map.iloc[
                  renumber_map_start_ix:renumber_map_end_ix
              ]
  
              # Add the length so no na-checking is required in the loading stage
              map_offset = (
                  end_batch_id - start_batch_id + 2
              ) - offsets_p.renumber_map_offsets.iloc[0]
              renumber_map_o = cudf.concat(
                  [
                      offsets_p.renumber_map_offsets + map_offset,
                      cudf.Series(
                          [len(renumber_map_p) + len(offsets_p) + 1],
                          dtype="int32",
                      ),
                  ]
              )
  
              renumber_offset_len = len(renumber_map_o)
              if (
                  renumber_offset_len
                  != end_batch_id - start_batch_id + 2
              ):
                  raise ValueError("Invalid batch id or renumber map")
  
              final_map_series = cudf.concat(
                  [
                      renumber_map_o,
                      renumber_map_p,
                  ],
                  ignore_index=True,
              )
  
              if len(final_map_series) > len(results_p):
                  # this should rarely happen and only occurs on small graphs/samples
                  # TODO remove the sort_index to improve performance on small graphs
                  final_map_series.name = "map"
                  results_p = results_p.join(
                      final_map_series, how="outer"
                  ).sort_index()
              else:
                  results_p["map"] = final_map_series
  
          full_output_path = os.path.join(
              output_path,
              f"batch={start_batch_id}-{end_batch_id}.parquet",
          )
  
          results_p.to_parquet(
              full_output_path,
              compression=None,
              index=False,
              force_nullable_schema=True,
          )
  
      return cudf.Series(dtype="int64")
  
file_path: cugraph/python/cugraph/cugraph/gnn/data_loading/bulk_sampler_io.py
imports: import os
  import cudf
  import cupy
  from math import ceil
  from pandas import isna
  from typing import Union, Optional, List
label_desc: Use image node_id 3 for calling a global function with example usage: _write_samples_to_parquet_coo(results, offsets, renumber_map, batches_per_partition, output_path, partition_info) and returns: cudf, cudf