class_docstr: Identifies feature skew in baseline and test examples.
  
  This PTransform returns a dict of PCollections containing:
    SKEW_RESULTS_KEY: Aggregated skew statistics (containing, e.g., mismatch
      count, baseline only, test only) for each feature; and
    SKEW_PAIRS_KEY: A sample of skewed example pairs (if sample_size is > 0).
    MATCH_STATS: A PCollection containing a single MatchStats proto.
    CONFUSION_KEY: (if configured) counts of paired feature values.
class_docstr_tok: ["Identifies","feature","skew","in","baseline","and","test","examples",".","This","PTransform","returns","a","dict","of","PCollections","containing",":","SKEW_RESULTS_KEY",":","Aggregated","skew","statistics","(","containing",",","e.g.",",","mismatch","count",",","baseline","only",",","test","only",")","for","each","feature",";","and","SKEW_PAIRS_KEY",":","A","sample","of","skewed","example","pairs","(","if","sample_size","is",">","0",")",".","MATCH_STATS",":","A","PCollection","containing","a","single","MatchStats","proto",".","CONFUSION_KEY",":","(","if","configured",")","counts","of","paired","feature","values","."]
code: def expand(
      self,
      pcollections: Tuple[
          beam.pvalue.PCollection, beam.pvalue.PCollection
      ],
  ) -> Dict[str, beam.pvalue.PCollection]:
      base_examples, test_examples = pcollections
      # Extract keyed base examples and counts of missing keys.
      keyed_base_examples_result = (
          base_examples
          | "ExtractBaseIdentifiers"
          >> beam.ParDo(
              _ExtractIdentifiers(
                  self._identifier_features, self._float_round_ndigits
              )
          ).with_outputs(_KEYED_EXAMPLE_KEY, _MISSING_IDS_KEY)
      )
  
      # Extract keyed test examples and counts of missing keys.
      keyed_test_examples_result = (
          test_examples
          | "ExtractTestIdentifiers"
          >> beam.ParDo(
              _ExtractIdentifiers(
                  self._identifier_features, self._float_round_ndigits
              )
          ).with_outputs(_KEYED_EXAMPLE_KEY, _MISSING_IDS_KEY)
      )
      (
          keyed_base_examples,
          missing_id_base_examples,
          keyed_test_examples,
          missing_id_test_examples,
      ) = _extract_extract_identifiers_result(
          keyed_base_examples_result, keyed_test_examples_result
      )
  
      outputs = [SKEW_RESULTS_KEY, SKEW_PAIRS_KEY, MATCH_STATS_KEY]
      if self._confusion_configs:
          outputs.append(CONFUSION_KEY)
      results = (
          {"base": keyed_base_examples, "test": keyed_test_examples}
          | "JoinExamples" >> beam.CoGroupByKey()
          | "ComputeSkew"
          >> beam.ParDo(
              _ComputeSkew(
                  self._features_to_ignore,
                  self._float_round_ndigits,
                  self._allow_duplicate_identifiers,
                  self._confusion_configs,
              )
          ).with_outputs(*outputs)
      )
      (
          results_skew_results,
          results_skew_pairs,
          results_match_stats,
          results_confusion_tuples,
      ) = _extract_compute_skew_result(results)
  
      outputs = {}
      # Merge skew results.
      skew_results = (
          results_skew_results
          | "MergeSkewResultsPerFeature"
          >> beam.CombinePerKey(  # pytype: disable=attribute-error
              _merge_feature_skew_results
          )
          | "DropKeys" >> beam.Values()
      )
      outputs[SKEW_RESULTS_KEY] = skew_results
  
      # Merge match stats.
      match_stats = (
          [
              results_match_stats,
              missing_id_test_examples,
              missing_id_base_examples,
          ]
          | "FlattenMatchStats" >> beam.Flatten()
          | "MergeMatchStats"
          >> beam.CombineGlobally(_MergeMatchStatsFn())
      )
      outputs[MATCH_STATS_KEY] = match_stats
  
      # Sample skew pairs.
      skew_pairs = (
          results_skew_pairs
          | "SampleSkewPairs"
          >> beam.combiners.Sample.FixedSizeGlobally(  # pytype: disable=attribute-error
              self._sample_size
          )
          # Sampling results in a pcollection with a single element consisting of
          # a list of the samples. Convert this to a pcollection of samples.
          | "Flatten" >> beam.FlatMap(lambda x: x)
      )
      outputs[SKEW_PAIRS_KEY] = skew_pairs
      if results_confusion_tuples is not None:
          confusion_counts = (
              results_confusion_tuples
              | "CountConfusion" >> beam.combiners.Count.PerElement()
              | "MakeConfusionProto"
              >> beam.Map(_confusion_count_to_proto)
          )
          outputs[CONFUSION_KEY] = confusion_counts
      return outputs
  
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 DetectFeatureSkewImpl obj's underlying member method code with example usage: obj.expand(pcollections) and returns: outputs