code: def test_single_phase_mixed_analyzer_run_once(self, use_tf_compat_v1):
      if not use_tf_compat_v1:
          tft_unit.skip_if_not_tf2("Tensorflow 2.x required.")
      span_0_key = analyzer_cache.DatasetKey("span-0")
      span_1_key = analyzer_cache.DatasetKey("span-1")
  
      def preprocessing_fn(inputs):
          _ = tft.bucketize(inputs["x"], 2, name="bucketize")
  
          return {
              "integerized_s": tft.compute_and_apply_vocabulary(
                  inputs["s"]
              ),
              "x_min": tft.min(inputs["x"], name="x")
              + tf.zeros_like(inputs["x"]),
              "x_mean": tft.mean(inputs["x"], name="x")
              + tf.zeros_like(inputs["x"]),
              "y_min": tft.min(inputs["y"], name="y")
              + tf.zeros_like(inputs["y"]),
              "y_mean": tft.mean(inputs["y"], name="y")
              + tf.zeros_like(inputs["y"]),
          }
  
      # Run AnalyzeAndTransform on some input data and compare with expected
      # output.
      input_data = [
          {"x": 12, "y": 1, "s": "d"},
          {"x": 10, "y": 1, "s": "c"},
      ]
      feature_spec = {
          "x": tf.io.FixedLenFeature([], tf.float32),
          "y": tf.io.FixedLenFeature([], tf.float32),
          "s": tf.io.FixedLenFeature([], tf.string),
      }
      input_data_dict = {
          span_0_key: [
              {
                  "x": -2,
                  "y": 1,
                  "s": "b",
              },
              {
                  "x": 4,
                  "y": -4,
                  "s": "b",
              },
          ],
          span_1_key: input_data,
      }
  
      span_0_size = 42
      cache_dict = {
          span_0_key: analyzer_cache.DatasetCache(
              {
                  _make_cache_key(
                      b"CacheableCombineAccumulate[x_1#mean_and_var]"
                  ): [b"[2.0, 1.0, 9.0, 0.0]"],
                  _make_cache_key(b"CacheableCombineAccumulate[x#x]"): [
                      b"[2.0, 4.0]"
                  ],
                  _make_cache_key(
                      b"CacheableCombineAccumulate[y_1#mean_and_var]"
                  ): [b"[2.0, -1.5, 6.25, 0.0]"],
                  _make_cache_key(b"CacheableCombineAccumulate[y#y]"): [
                      b"[4.0, 1.0]"
                  ],
              },
              analyzer_cache.DatasetCacheMetadata(span_0_size),
          ),
          span_1_key: analyzer_cache.DatasetCache({}, None),
      }
  
      expected_transformed = [
          {
              "x_mean": 6.0,
              "x_min": -2.0,
              "y_mean": -0.25,
              "y_min": -4.0,
              "integerized_s": 1,
          },
          {
              "x_mean": 6.0,
              "x_min": -2.0,
              "y_mean": -0.25,
              "y_min": -4.0,
              "integerized_s": 2,
          },
      ]
  
      run_result = self._run_pipeline(
          feature_spec,
          input_data_dict,
          preprocessing_fn,
          cache_dict=cache_dict,
          datasets_to_transform=[span_1_key],
          expected_transform_data=expected_transformed,
          transform_fn_output_dir=os.path.join(
              self.base_test_dir, "transform_fn"
          ),
          use_tf_compat_v1=use_tf_compat_v1,
      )
  
      # The output cache should not have entries for the cache that is present
      # in the input cache.
      self.assertEqual(
          len(run_result.cache_output[span_0_key].cache_dict),
          len(run_result.cache_output[span_1_key].cache_dict) - 4,
      )
  
      metrics = run_result.metrics
      # 4 from analyzing 2 spans, and 2 from transform.
      self.assertMetricsCounterEqual(metrics, "num_instances", 6)
      self.assertMetricsCounterEqual(
          metrics, "cache_entries_decoded", 4
      )
      self.assertMetricsCounterEqual(
          metrics, "cache_entries_encoded", 8
      )
      self.assertMetricsCounterEqual(
          metrics,
          "saved_models_created",
          _SINGLE_PHASE_NUM_SAVED_MODELS,
      )
      self.assertMetricsCounterEqual(
          metrics, "num_packed_accumulate_combiners", 1
      )
      self.assertMetricsCounterEqual(
          metrics, "num_packed_merge_combiners", 1
      )
      # All datasets were processed even though some of the analyzers were covered
      # by cache.
      self.assertMetricsCounterEqual(
          metrics, "analysis_input_bytes_from_cache", 0
      )
  
imports: import functools
  import os
  import struct
  from typing import Callable, Mapping, List
  import uuid
  import apache_beam
  from apache_beam.testing import util
  import numpy
  import tensorflow
  import tensorflow_transform
  from tensorflow_transform import analyzer_nodes
  from tensorflow_transform import common_types
  from tensorflow_transform import impl_helper
  from tensorflow_transform import nodes
  import tensorflow_transform.beam
  from tensorflow_transform.beam import analysis_graph_builder
  from tensorflow_transform.beam import analyzer_cache
  from tensorflow_transform.beam import tft_unit
  from tensorflow_transform.tf_metadata import dataset_metadata
  from tfx_bsl.types import tfx_namedtuple
label_desc: Use image node_id 7 for calling the CachedImplTest obj's underlying member method code with example usage: obj.test_single_phase_mixed_analyzer_run_once(use_tf_compat_v1) and returns: dict