code: def _write_samples_to_parquet_csr(
      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 CSR/CSC 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")
  
      # Additional check to skip dummy partitions required for CSR format.
      if isna(offsets.batch_id.iloc[0]):
          return cudf.Series(dtype="int64")
  
      # Output:
      # major_offsets - CSR/CSC row/col pointers
      # minors - CSR/CSC col/row indices
      # edge id - edge ids (same shape as minors)
      # edge type - edge types (same shape as minors)
      # weight - edge weight (same shape as minors)
      # renumber map - the original vertex ids
      # renumber map offsets - start/end of the map for each batch
      #                        (only 1 per batch b/c of framework
      #                         stipulations making this legal)
      # label-hop offsets - indicate the start/end of each hop
      #                     for each batch
  
      batch_ids = offsets.batch_id
      label_hop_offsets = offsets.offsets
      renumber_map_offsets = offsets.renumber_map_offsets
      del offsets
  
      batch_ids.dropna(inplace=True)
      label_hop_offsets.dropna(inplace=True)
      renumber_map_offsets.dropna(inplace=True)
  
      major_offsets_array = results.major_offsets
      results.drop(columns="major_offsets", inplace=True)
      major_offsets_array.dropna(inplace=True)
      major_offsets_array = major_offsets_array.values
  
      minors_array = results.minors
      results.drop(columns="minors", inplace=True)
      minors_array.dropna(inplace=True)
      minors_array = minors_array.values
  
      weight_array = results.weight
      results.drop(columns="weight", inplace=True)
      weight_array.dropna(inplace=True)
      weight_array = (
          cupy.array([], dtype="float32")
          if weight_array.empty
          else weight_array.values
      )
  
      edge_id_array = results.edge_id
      results.drop(columns="edge_id", inplace=True)
      edge_id_array.dropna(inplace=True)
      edge_id_array = (
          cupy.array([], dtype="int64")
          if edge_id_array.empty
          else edge_id_array.values
      )
  
      edge_type_array = results.edge_type
      results.drop(columns="edge_type", inplace=True)
      edge_type_array.dropna(inplace=True)
      edge_type_array = (
          cupy.array([], dtype="int32")
          if edge_type_array.empty
          else edge_type_array.values
      )
  
      del results
  
      offsets_length = len(label_hop_offsets) - 1
      if offsets_length % len(batch_ids) != 0:
          raise ValueError("Invalid hop offsets")
      fanout_length = int(offsets_length / len(batch_ids))
  
      for p in range(
          0, int(ceil(len(batch_ids) / batches_per_partition))
      ):
          partition_start = p * (batches_per_partition)
          partition_end = (p + 1) * (batches_per_partition)
  
          label_hop_offsets_current_partition = label_hop_offsets.iloc[
              partition_start
              * fanout_length : partition_end
              * fanout_length
              + 1
          ].reset_index(drop=True)
          label_hop_offsets_current_partition.name = "label_hop_offsets"
  
          batch_ids_current_partition = batch_ids.iloc[
              partition_start:partition_end
          ]
  
          (
              major_offsets_start,
              major_offsets_end,
          ) = label_hop_offsets_current_partition.iloc[
              [0, -1]
          ].values  # legal since offsets has the 1 extra offset
          results_start, results_end = major_offsets_array[
              [major_offsets_start, major_offsets_end]
          ]  # avoid d2h copy
  
          # no need to use end batch id, just ensure the batch is labeled correctly
          start_batch_id = batch_ids_current_partition.iloc[0]
          # end_batch_id = batch_ids_current_partition.iloc[-1]
  
          # create the renumber map offsets
          renumber_map_offsets_current_partition = (
              renumber_map_offsets.iloc[
                  partition_start : partition_end + 1
              ].reset_index(drop=True)
          )
          renumber_map_offsets_current_partition.name = (
              "renumber_map_offsets"
          )
  
          (
              renumber_map_start,
              renumber_map_end,
          ) = renumber_map_offsets_current_partition.iloc[
              [0, -1]
          ].values  # avoid d2h copy
  
          results_current_partition = create_df_from_disjoint_series(
              [
                  cudf.Series(
                      minors_array[results_start:results_end],
                      name="minors",
                  ),
                  cudf.Series(
                      renumber_map.map.values[
                          renumber_map_start:renumber_map_end
                      ],
                      name="map",
                  ),
                  label_hop_offsets_current_partition,
                  cudf.Series(
                      major_offsets_array[
                          major_offsets_start : major_offsets_end + 1
                      ],
                      name="major_offsets",
                  ),
                  cudf.Series(
                      weight_array[results_start:results_end],
                      name="weight",
                  ),
                  cudf.Series(
                      edge_id_array[results_start:results_end],
                      name="edge_id",
                  ),
                  cudf.Series(
                      edge_type_array[results_start:results_end],
                      name="edge_type",
                  ),
                  renumber_map_offsets_current_partition,
              ]
          )
  
          end_batch_id = (
              start_batch_id + len(batch_ids_current_partition) - 1
          )
          filename = f"batch={start_batch_id}-{end_batch_id}.parquet"
          full_output_path = os.path.join(output_path, filename)
  
          results_current_partition.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 2 for calling a global function with example usage: _write_samples_to_parquet_csr(results, offsets, renumber_map, batches_per_partition, output_path, partition_info) and returns: cudf, cudf, cudf