code: def _filter_index_sort(raw_rating_path, cache_path):
      # type: (str, str, bool) -> (dict, bool)
      """Read in data CSV, and output structured data.
  
      This function reads in the raw CSV of positive items, and performs three
      preprocessing transformations:
  
      1)  Filter out all users who have not rated at least a certain number
          of items. (Typically 20 items)
  
      2)  Zero index the users and items such that the largest user_id is
          `num_users - 1` and the largest item_id is `num_items - 1`
  
      3)  Sort the dataframe by user_id, with timestamp as a secondary sort key.
          This allows the dataframe to be sliced by user in-place, and for the last
          item to be selected simply by calling the `-1` index of a user's slice.
  
      While all of these transformations are performed by Pandas (and are therefore
      single-threaded), they only take ~2 minutes, and the overhead to apply a
      MapReduce pattern to parallel process the dataset adds significant complexity
      for no computational gain. For a larger dataset parallelizing this
      preprocessing could yield speedups. (Also, this preprocessing step is only
      performed once for an entire run.
  
      Args:
        raw_rating_path: The path to the CSV which contains the raw dataset.
        cache_path: The path to the file where results of this function are saved.
  
      Returns:
        A filtered, zero-index remapped, sorted dataframe, a dict mapping raw user
        IDs to regularized user IDs, and a dict mapping raw item IDs to regularized
        item IDs.
      """
      valid_cache = tf.io.gfile.exists(cache_path)
      if valid_cache:
          with tf.io.gfile.GFile(cache_path, "rb") as f:
              cached_data = pickle.load(f)
  
          # (nnigania)disabled this check as the dataset is not expected to change
          # cache_age = time.time() - cached_data.get("create_time", 0)
          # if cache_age > rconst.CACHE_INVALIDATION_SEC:
          #   valid_cache = False
  
          for key in _EXPECTED_CACHE_KEYS:
              if key not in cached_data:
                  valid_cache = False
  
          if not valid_cache:
              logging.info("Removing stale raw data cache file.")
              tf.io.gfile.remove(cache_path)
  
      if valid_cache:
          data = cached_data
      else:
          with tf.io.gfile.GFile(raw_rating_path) as f:
              df = pd.read_csv(f)
  
          # Get the info of users who have more than 20 ratings on items
          grouped = df.groupby(movielens.USER_COLUMN)
          df = grouped.filter(
              lambda x: len(x) >= rconst.MIN_NUM_RATINGS
          )  # type: pd.DataFrame
  
          original_users = df[movielens.USER_COLUMN].unique()
          original_items = df[movielens.ITEM_COLUMN].unique()
  
          # Map the ids of user and item to 0 based index for following
          # processing
          logging.info("Generating user_map and item_map...")
          user_map = {
              user: index for index, user in enumerate(original_users)
          }
          item_map = {
              item: index for index, item in enumerate(original_items)
          }
  
          df[movielens.USER_COLUMN] = df[movielens.USER_COLUMN].apply(
              lambda user: user_map[user]
          )
          df[movielens.ITEM_COLUMN] = df[movielens.ITEM_COLUMN].apply(
              lambda item: item_map[item]
          )
  
          num_users = len(original_users)
          num_items = len(original_items)
  
          mlperf_helper.ncf_print(
              key=mlperf_helper.TAGS.PREPROC_HP_NUM_EVAL,
              value=rconst.NUM_EVAL_NEGATIVES,
          )
  
          assert num_users <= np.iinfo(rconst.USER_DTYPE).max
          assert num_items <= np.iinfo(rconst.ITEM_DTYPE).max
          assert df[movielens.USER_COLUMN].max() == num_users - 1
          assert df[movielens.ITEM_COLUMN].max() == num_items - 1
  
          # This sort is used to shard the dataframe by user, and later to select
          # the last item for a user to be used in validation.
          logging.info("Sorting by user, timestamp...")
  
          # This sort is equivalent to
          #   df.sort_values([movielens.USER_COLUMN, movielens.TIMESTAMP_COLUMN],
          #   inplace=True)
          # except that the order of items with the same user and timestamp are
          # sometimes different. For some reason, this sort results in a better
          # hit-rate during evaluation, matching the performance of the MLPerf
          # reference implementation.
          df.sort_values(by=movielens.TIMESTAMP_COLUMN, inplace=True)
          df.sort_values(
              [movielens.USER_COLUMN, movielens.TIMESTAMP_COLUMN],
              inplace=True,
              kind="mergesort",
          )
  
          # The dataframe does not reconstruct indices in the sort or filter
          # steps.
          df = df.reset_index()
  
          grouped = df.groupby(movielens.USER_COLUMN, group_keys=False)
          eval_df, train_df = grouped.tail(1), grouped.apply(
              lambda x: x.iloc[:-1]
          )
  
          data = {
              rconst.TRAIN_USER_KEY: train_df[
                  movielens.USER_COLUMN
              ].values.astype(rconst.USER_DTYPE),
              rconst.TRAIN_ITEM_KEY: train_df[
                  movielens.ITEM_COLUMN
              ].values.astype(rconst.ITEM_DTYPE),
              rconst.EVAL_USER_KEY: eval_df[
                  movielens.USER_COLUMN
              ].values.astype(rconst.USER_DTYPE),
              rconst.EVAL_ITEM_KEY: eval_df[
                  movielens.ITEM_COLUMN
              ].values.astype(rconst.ITEM_DTYPE),
              rconst.USER_MAP: user_map,
              rconst.ITEM_MAP: item_map,
              "create_time": time.time(),
          }
  
          logging.info("Writing raw data cache.")
          with tf.io.gfile.GFile(cache_path, "wb") as f:
              pickle.dump(data, f, protocol=pickle.HIGHEST_PROTOCOL)
  
      # TODO(robieta): MLPerf cache clear.
      return data, valid_cache
  
file_path: autodist/examples/benchmark/utils/recommendation/data_preprocessing.py
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  import os
  import pickle
  import time
  import timeit
  import typing
  import numpy
  import pandas
  import tensorflow
  from absl import logging
  from utils.recommendation import constants
  from utils.recommendation import data_pipeline
  from utils.recommendation import movielens
  from utils.logs import mlperf_helper
label_desc: Use image node_id 1 for calling a global function with example usage: _filter_index_sort(raw_rating_path, cache_path) and returns: data,  valid_cache