class_docstr: A DoFn that performs per-slice key fanout prior to computing aggregates.
class_docstr_tok: ["A","DoFn","that","performs","per-slice","key","fanout","prior","to","computing","aggregates","."]
code: def process(
      self, element: types.Extracts
  ) -> List[Tuple[SliceKeyType, types.Extracts]]:
      key_filter_fn = self._key_filter_fn  # Local cache.
      filtered = {k: v for k, v in element.items() if key_filter_fn(k)}
      slice_keys = element.get(constants.SLICE_KEY_TYPES_KEY)
      # The query based evaluator will group slices from multiple examples, so we
      # deduplicate to avoid overcounting. Depending on whether the rows within a
      # batch have a variable or fixed length, either a VarLenTensorValue or a 2D
      # np.ndarray will be created.
      if isinstance(slice_keys, types.VarLenTensorValue):
          slice_keys = slice_keys.values
      elif (
          isinstance(slice_keys, np.ndarray)
          and len(slice_keys.shape) == 2
      ):
          slice_keys = slice_keys.flatten()
      result = [(slice_key, filtered) for slice_key in set(slice_keys)]
      self._num_slices_generated_per_instance.update(len(result))
      self._post_slice_num_instances.inc(len(result))
      return result
  
imports: import itertools
  from typing import Any, Callable, Dict, Generator, Iterable, List, NamedTuple, Optional, Tuple, Union
  import apache_beam
  import numpy
  import tensorflow
  from tensorflow_model_analysis import constants
  from tensorflow_model_analysis.api import types
  from tensorflow_model_analysis.proto import config_pb2
  from tensorflow_model_analysis.proto import metrics_for_slice_pb2
  from tensorflow_model_analysis.slicer import slice_accessor
label_desc: Use image node_id 2 for calling the _FanoutSlicesDoFn obj's underlying member method code with example usage: obj.process(element) and returns: result