code: def get_feature_value_slicer(
      features: Dict[types.FeatureName, Optional[_ValueType]]
  ) -> types.SliceFunction:
      """Returns a function that generates sliced record batches for a given one.
  
      The returned function returns sliced record batches based on the combination
      of all features specified in `features`. To slice on features separately (
      e.g., slice on age feature and separately slice on interests feature), you
      must use separate slice functions.
  
      Examples:
      # Slice on each value of the specified features.
      slice_fn = get_feature_value_slicer(
          features={'age': None, 'interests': None})
  
      # Slice on a specified feature value.
      slice_fn = get_feature_value_slicer(features={'interests': ['dogs']})
  
      # Slice on each value of one feature and a specified value of another.
      slice_fn = get_feature_value_slicer(
          features={'fruits': None, 'numbers': [1]})
  
      Args:
        features: A mapping of features to an optional iterable of values that the
          returned function will slice on. If values is None for a feature, then the
          slice keys will reflect each distinct value found for that feature in the
          input record batch. If values are specified for a feature, then the slice
          keys will reflect only those values for the feature, if found in the input
          record batch. Values must be an iterable of strings or integers.
  
      Returns:
        A function that takes as input a single record batch and returns a list of
        sliced record batches (slice_key, record_batch).
  
      Raises:
        TypeError: If feature values are not specified in an iterable.
        NotImplementedError: If a value of a type other than string or integer is
          specified in the values iterable in `features`.
      """
      for values in features.values():
          if values is not None:
              if not isinstance(values, abc.Iterable):
                  raise TypeError(
                      "Feature values must be specified in an iterable."
                  )
              for value in values:
                  if not isinstance(
                      value, (six.string_types, six.binary_type)
                  ) and not isinstance(value, int):
                      raise NotImplementedError(
                          "Only string and int values are supported as the slice value."
                      )
      # Extract the unique slice values per feature.
      for feature_name in features:
          if features[feature_name] is not None:
              features[feature_name] = set(features[feature_name])
  
      def feature_value_slicer(
          record_batch: pa.RecordBatch,
      ) -> Iterable[types.SlicedRecordBatch]:
          """A function that generates sliced record batches.
  
          The naive approach of doing this would be to iterate each row, identify
          slice keys for the row and keep track of index ranges for each slice key.
          And then generate an arrow record batch for each slice key based on the
          index ranges. This would be expensive as we are identifying the slice keys
          for each row individually and we would have to loop over the feature values
          including crossing them when we have to slice on multiple features. The
          current approach generates the slice keys for a batch by performing joins
          over indices of individual features. And then groups the joined record batch
          by slice key to get the row indices corresponding to a slice.
  
          Args:
            record_batch: Arrow RecordBatch.
  
          Yields:
            Sliced record batch (slice_key, record_batch) where record_batch contains
            the rows corresponding to a slice.
          """
          per_feature_parent_indices = []
          for feature_name, values in six.iteritems(features):
              feature_array = arrow_util.get_column(
                  record_batch, feature_name, missing_ok=True
              )
              # If the feature name does not appear in the schema for this record batch,
              # drop it from the set of sliced features.
              if feature_array is None:
                  continue
  
              # convert values from list[str] to list[int] if the feature type
              # is integer.
              if values is not None:
                  feature_type = (
                      stats_util.get_feature_type_from_arrow_type(
                          types.FeaturePath([feature_name]),
                          feature_array.type,
                      )
                  )
                  if (
                      feature_type
                      == statistics_pb2.FeatureNameStatistics.INT
                  ):
                      try:
                          values = [int(value) for value in values]
                      except ValueError as e:
                          raise ValueError(
                              "The feature to slice on has integer values but "
                              "the provided slice values are not valid integers."
                          ) from e
  
              (
                  flattened,
                  value_parent_indices,
              ) = array_util.flatten_nested(feature_array, True)
              non_missing_values = np.asarray(flattened)
              # Create dataframe with feature value and parent index.
              df = pd.DataFrame(
                  {
                      feature_name: non_missing_values,
                      _PARENT_INDEX_COLUMN: value_parent_indices,
                  }
              )
              df.drop_duplicates(inplace=True)
              # Filter based on slice values
              if values is not None:
                  df = df.loc[df[feature_name].isin(values)]
              per_feature_parent_indices.append(df)
          # If there are no features to slice on, yield no output.
          # TODO(b/200081813): Produce output with an appropriate placeholder key.
          if not per_feature_parent_indices:
              return
          # Join dataframes based on parent indices.
          # Note that we want the parent indices per slice key to be sorted in the
          # merged dataframe. The individual dataframes have the parent indices in
          # sorted order. We use "inner" join type to preserve the order of the left
          # keys (also note that same parent index rows would be consecutive). Hence
          # we expect the merged dataframe to have sorted parent indices per
          # slice key.
          merged_df = functools.reduce(
              lambda base, update: pd.merge(
                  base,
                  update,
                  how="inner",  # pylint: disable=g-long-lambda
                  on=_PARENT_INDEX_COLUMN,
              ),
              per_feature_parent_indices,
          )
  
          # Construct a new column in the merged dataframe with the slice keys.
          merged_df[_SLICE_KEY_COLUMN] = ""
          index = 0
          for col_name in sorted(merged_df.columns):
              if col_name in [_PARENT_INDEX_COLUMN, _SLICE_KEY_COLUMN]:
                  continue
              feature_value_part = merged_df[col_name].apply(
                  _to_slice_key
              )
              if feature_value_part.empty:
                  feature_value_part = feature_value_part.astype(
                      pd.StringDtype()
                  )
              slice_key_col = (
                  _to_slice_key(col_name) + "_" + feature_value_part
              )
              if index == 0:
                  merged_df[_SLICE_KEY_COLUMN] = slice_key_col
                  index += 1
              else:
                  merged_df[_SLICE_KEY_COLUMN] += "_" + slice_key_col
  
          # Since the parent indices are sorted per slice key, the groupby would
          # preserve the sorted order within each group.
          per_slice_parent_indices = merged_df.groupby(
              _SLICE_KEY_COLUMN, sort=False
          )[_PARENT_INDEX_COLUMN]
          for slice_key, parent_indices in per_slice_parent_indices:
              yield (
                  slice_key,
                  table_util.RecordBatchTake(
                      record_batch, pa.array(parent_indices.to_numpy())
                  ),
              )
  
      return feature_value_slicer
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  import collections
  from collections import abc
  import functools
  import logging
  from typing import Any, Dict, Iterable, List, Optional, Text, Union, Tuple
  import apache_beam
  import numpy
  import pandas
  import pandas.core.computation.expressions
  import pyarrow
  import six
  from tensorflow_data_validation import constants
  from tensorflow_data_validation import types
  from tensorflow_data_validation.arrow import arrow_util
  from tensorflow_data_validation.utils import stats_util
  from tfx_bsl.arrow import array_util
  from tfx_bsl.arrow import sql_util
  from tfx_bsl.arrow import table_util
  from tfx_bsl.public.proto import slicing_spec_pb2
  from tensorflow_metadata.proto.v0 import statistics_pb2
label_desc: Use image node_id 2 for calling a global function with example usage: get_feature_value_slicer(features) and returns: feature_value_slicer