code: def test_cached_ptransform_analyzer(self, use_tf_compat_v1):
      if not use_tf_compat_v1:
          tft_unit.skip_if_not_tf2("Tensorflow 2.x required.")
  
      class _AnalyzerMakeAccumulators(beam.PTransform):
          def expand(self, pcoll):
              input_sum = (
                  pcoll
                  | beam.FlatMap(sum)
                  | "ReduceSum" >> beam.CombineGlobally(sum)
              )
              size = (
                  pcoll
                  | beam.Map(np.size)
                  | "ReduceCount" >> beam.CombineGlobally(sum)
              )
  
              return (
                  pcoll.pipeline
                  | beam.Create([None])
                  | beam.Map(
                      lambda _, a, b: (a, b),  # pyformat: disable
                      beam.pvalue.AsSingleton(input_sum),
                      beam.pvalue.AsSingleton(size),
                  )
              )
  
      class _AnalyzerMergeAccumulators(beam.PTransform):
          def expand(self, pcoll):
              def merge(x):
                  zipped = list(zip(*x))
                  assert len(zipped) == 2, zipped
                  return sum(zipped[0]), sum(zipped[1])
  
              return pcoll | beam.CombineGlobally(merge)
  
      class _AnalyzerExtractOutput(beam.PTransform):
          def expand(self, pcoll):
              return pcoll | beam.Map(lambda p: p[0] / p[1])
  
      analyzer = tft.experimental.CacheablePTransformAnalyzer(
          make_accumulators_ptransform=_AnalyzerMakeAccumulators(),
          merge_accumulators_ptransform=_AnalyzerMergeAccumulators(),
          extract_output_ptransform=_AnalyzerExtractOutput(),
          cache_coder=tft.experimental.SimpleJsonPTransformAnalyzerCacheCoder(),
      )
  
      def preprocessing_fn(inputs):
          y = tft.experimental.ptransform_analyzer(
              [inputs["x"]], analyzer, [tf.int64], [[]]
          )
          return {"y": tf.zeros_like(inputs["x"]) + y}
  
      feature_spec = {"x": tf.io.FixedLenFeature([], tf.int64)}
      span_0_key = analyzer_cache.DatasetKey("span-0")
      input_data_dict = {span_0_key: [{"x": x} for x in range(7)]}
      expected_cache_dict = {
          span_0_key: {
              _make_cache_key(
                  b"PTransform[ptransform#local_merge_accumulators]"
              ): [b"[21, 7]"],
          },
      }
      expected_transformed_data = [{"y": 3} for _ in range(7)]
      first_run_result = self._run_pipeline(
          feature_spec,
          input_data_dict,
          preprocessing_fn,
          datasets_to_transform=[span_0_key],
          expected_transform_data=expected_transformed_data,
          expected_cache=expected_cache_dict,
          use_tf_compat_v1=use_tf_compat_v1,
      )
      metrics = first_run_result.metrics
      # Incremented for both analysis and transform (7 * 2).
      self.assertMetricsCounterEqual(metrics, "num_instances", 14)
      self.assertMetricsCounterEqual(
          metrics, "cache_entries_decoded", 0
      )
      self.assertMetricsCounterEqual(
          metrics, "cache_entries_encoded", 1
      )
      self.assertMetricsCounterEqual(
          metrics, "analysis_input_bytes_from_cache", 0
      )
  
      first_run_result = self._run_pipeline(
          feature_spec,
          input_data_dict,
          preprocessing_fn,
          should_read_cache=True,
          datasets_to_transform=[span_0_key],
          expected_transform_data=expected_transformed_data,
          expected_cache={},
          use_tf_compat_v1=use_tf_compat_v1,
      )
      metrics = first_run_result.metrics
      # This time analysis is skipped due to cache, only transform dataset counts.
      self.assertMetricsCounterEqual(metrics, "num_instances", 7)
      self.assertMetricsCounterEqual(
          metrics, "cache_entries_decoded", 1
      )
      self.assertMetricsCounterEqual(
          metrics, "cache_entries_encoded", 0
      )
      self.assertMetricsCounterGreater(
          metrics, "analysis_input_bytes_from_cache", 100
      )
  
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 11 for calling the CachedImplTest obj's underlying member method code with example usage: obj.test_cached_ptransform_analyzer(use_tf_compat_v1) and returns: dict, unknown, unknown, unknown, sum,  sum