class_docstr: DoFn that computes skew for each pair of examples.
class_docstr_tok: ["DoFn","that","computes","skew","for","each","pair","of","examples","."]
code: def process(
      self, element: Tuple[str, Dict[str, Iterable[Any]]]
  ) -> Iterable[_PairOrFeatureSkew]:
      (_, examples) = element
      base_examples = list(examples.get(_BASELINE_KEY))
      test_examples = list(examples.get(_TEST_KEY))
  
      match_stats = _make_match_stats_counter(
          len(base_examples),
          len(test_examples),
          1,
          0 if base_examples else 1,
          0 if test_examples else 1,
          len(base_examples) * len(test_examples),
          1 if len(base_examples) > 1 or len(test_examples) > 1 else 0,
      )
      yield beam.pvalue.TaggedOutput(MATCH_STATS_KEY, match_stats)
      self._ids_counter.inc(1)
      self._pairs_counter.inc(len(base_examples) * len(test_examples))
      if not self._allow_duplicate_identifiers:
          if len(base_examples) > 1 or len(test_examples) > 1:
              self._skipped_duplicate_identifiers_counter.inc(1)
              return
      if base_examples and test_examples:
          for base_example in base_examples:
              for test_example in test_examples:
                  result, is_skewed = _compute_skew_for_examples(
                      base_example,
                      test_example,
                      self._features_to_ignore,
                      self._float_round_ndigits,
                  )
                  if is_skewed:
                      skew_pair = _construct_skew_pair(
                          result, base_example, test_example
                      )
                      yield beam.pvalue.TaggedOutput(
                          SKEW_PAIRS_KEY, skew_pair
                      )
                  for each in result:
                      yield beam.pvalue.TaggedOutput(
                          SKEW_RESULTS_KEY, each
                      )
                  if self._confusion_configs is not None:
                      for pair in _yield_confusion_pairs(
                          base_example,
                          test_example,
                          self._confusion_configs,
                      ):
                          yield beam.pvalue.TaggedOutput(
                              CONFUSION_KEY, pair
                          )
  
imports: from typing import Any, Dict, Iterable, Iterator, List, Optional, Tuple, Union
  import apache_beam
  import farmhash
  import numpy
  import tensorflow
  from tensorflow_data_validation import constants
  from tensorflow_data_validation import types
  from tensorflow_data_validation.utils import artifacts_io_impl
  from tensorflow_data_validation.skew.protos import feature_skew_results_pb2
label_desc: Use image node_id 2 for calling the _ComputeSkew obj's underlying member method code with example usage: obj.process(element) without return types