code: def _index_columns(
      piece_info,
      dataset_url,
      partitions,
      indexers,
      schema,
      hdfs_driver="libhdfs3",
  ):
      """
      Function build indexes for  dataset piece described in piece_info
      :param piece_info: description of dataset piece
      :param dataset_url: dataset location
      :param partitions: dataset partitions
      :param indexers: list of indexer objects
      :param schema: dataset schema
      :param hdfs_driver: A string denoting the hdfs driver to use (if using a dataset on hdfs). Current choices are
          libhdfs (java through JNI) or libhdfs3 (C++)
      :return: list of indexers containing index data
      """
      # Resolver in executor context will get hadoop config from environment
      resolver = FilesystemResolver(
          dataset_url, hdfs_driver=hdfs_driver
      )
      fs = resolver.filesystem()
  
      # Create pyarrow piece
      piece = pq.ParquetDatasetPiece(
          piece_info.path,
          open_file_func=fs.open,
          row_group=piece_info.row_group,
          partition_keys=piece_info.partition_keys,
      )
  
      # Collect column names needed for indexing
      column_names = set()
      for indexer in indexers:
          column_names.update(indexer.column_names)
  
      # Read columns needed for indexing
      column_rows = (
          piece.read(columns=list(column_names), partitions=partitions)
          .to_pandas()
          .to_dict("records")
      )
  
      # Decode columns values
      decoded_rows = [
          utils.decode_row(row, schema) for row in column_rows
      ]
      if not decoded_rows:
          raise ValueError(
              "Cannot build index with empty decoded_rows, columns: {}, partitions: {}".format(
                  column_names, partitions
              )
          )
  
      # Index columns values
      for indexer in indexers:
          indexer.build_index(decoded_rows, piece_info.piece_index)
  
      # Indexer objects contain index data, it will be consolidated on reduce phace
      return indexers
  
file_path: petastorm/petastorm/etl/rowgroup_indexing.py
imports: import logging
  import time
  from collections import namedtuple
  from pyarrow import parquet
  from six.moves import cPickle
  from six.moves import range
  from petastorm import utils
  from petastorm.etl import dataset_metadata
  from petastorm.etl.legacy import depickle_legacy_package_name_compatible
  from petastorm.fs_utils import FilesystemResolver
label_desc: Use image node_id 2 for calling a global function with example usage: _index_columns(piece_info, dataset_url, partitions, indexers, schema, hdfs_driver) and returns: indexers