def _apply_analyzer(ptransform: Union[_BeamPTransform,
                                      CacheablePTransformAnalyzer],
                    *tensor_inputs: common_types.TensorType,
                    **analyzer_def_kwargs: Any) -> Tuple[tf.Tensor, ...]:
  """Applies the analyzer over the whole dataset.

  Args:
    ptransform: A class inheriting from analyzer_nodes.AnalyzerDef or
      CacheablePTransformAnalyzer that should be applied.
    *tensor_inputs: A list of input `Tensor`s, `SparseTensor`s, or
      `RaggedTensor`s.
    **analyzer_def_kwargs: KW arguments to use when constructing
      analyzer_def_cls.

  Returns:
    A list of `Tensor`s representing the values of the analysis result.
  """
  input_values_node = analyzer_nodes.get_input_tensors_value_nodes(
      tensor_inputs)
  if isinstance(ptransform, CacheablePTransformAnalyzer):
    with tf.compat.v1.name_scope('make_accumulators'):
      make_accumulators_value_node = nodes.apply_multi_output_operation(
          analyzer_nodes.PTransform,
          input_values_node,
          ptransform=ptransform.make_accumulators_ptransform,
          is_partitionable=True,
          **analyzer_def_kwargs)
    with tf.compat.v1.name_scope('local_merge_accumulators'):
      cached_value_nodes = nodes.apply_multi_output_operation(
          analyzer_nodes.PTransform,
          *make_accumulators_value_node,
          ptransform=ptransform.merge_accumulators_ptransform,
          is_partitionable=True,
          cache_coder=ptransform.cache_coder,
          **analyzer_def_kwargs)
    with tf.compat.v1.name_scope('global_merge_accumulators'):
      merge_output_value_nodes = nodes.apply_multi_output_operation(
          analyzer_nodes.PTransform,
          *cached_value_nodes,
          ptransform=ptransform.merge_accumulators_ptransform,
          is_partitionable=False,
          **analyzer_def_kwargs)
    with tf.compat.v1.name_scope('extract_output'):
      output_value_nodes = nodes.apply_multi_output_operation(
          analyzer_nodes.PTransform,
          *merge_output_value_nodes,
          ptransform=ptransform.extract_output_ptransform,
          is_partitionable=False,
          **analyzer_def_kwargs)
  else:
    output_value_nodes = nodes.apply_multi_output_operation(
        analyzer_nodes.PTransform,
        input_values_node,
        ptransform=ptransform,
        is_partitionable=False,
        **analyzer_def_kwargs)
  return tuple(map(analyzer_nodes.wrap_as_tensor, output_value_nodes))
Applies the analyzer over the whole dataset.

  Args:
    ptransform: A class inheriting from analyzer_nodes.AnalyzerDef or
      CacheablePTransformAnalyzer that should be applied.
    *tensor_inputs: A list of input `Tensor`s, `SparseTensor`s, or
      `RaggedTensor`s.
    **analyzer_def_kwargs: KW arguments to use when constructing
      analyzer_def_cls.

  Returns:
    A list of `Tensor`s representing the values of the analysis result.
  
tensorflow_transform/experimental/analyzers.py