code: def ptransform_analyzer(
      inputs: Collection[tf.Tensor],
      ptransform: Union[_BeamPTransform, CacheablePTransformAnalyzer],
      output_dtypes: Collection[tf.dtypes.DType],
      output_shapes: Collection[List[int]],
      output_asset_default_values: Optional[
          Collection[Optional[bytes]]
      ] = None,
      name: Optional[str] = None,
  ):
      # pylint: disable=line-too-long
      """Applies a user-provided PTransform over the whole dataset.
  
      WARNING: This is experimental.
  
      Note that in order to have asset files copied correctly, any outputs that
      represent asset filenames must be added to the `tf.GraphKeys.ASSET_FILEPATHS`
      collection by the caller if using Transform's APIs in compat v1 mode.
  
      Example:
  
      >>> class MeanPerKey(beam.PTransform):
      ...   def expand(self, pcoll: beam.PCollection[Tuple[np.ndarray, np.ndarray]]) -> Tuple[beam.PCollection[np.ndarray], beam.PCollection[np.ndarray]]:
      ...     def extract_output(key_value_pairs):
      ...       keys, values = zip(*key_value_pairs)
      ...       return [beam.TaggedOutput('keys', keys),
      ...               beam.TaggedOutput('values', values)]
      ...     return tuple(
      ...         pcoll
      ...         | 'ZipAndFlatten' >> beam.FlatMap(lambda batches: list(zip(*batches)))
      ...         | 'MeanPerKey' >> beam.CombinePerKey(beam.combiners.MeanCombineFn())
      ...         | 'ToList' >> beam.combiners.ToList()
      ...         | 'Extract' >> beam.FlatMap(extract_output).with_outputs(
      ...             'keys', 'values'))
      >>> def preprocessing_fn(inputs):
      ...   outputs = tft.experimental.ptransform_analyzer(
      ...       inputs=[inputs['s'], inputs['x']],
      ...       ptransform=MeanPerKey(),
      ...       output_dtypes=[tf.string, tf.float32],
      ...       output_shapes=[[2], [2]])
      ...   (keys, means) = outputs
      ...   mean_a = tf.reshape(tf.gather(means, tf.where(keys == 'a')), [])
      ...   return { 'x/mean_a': inputs['x'] / mean_a }
      >>> raw_data = [dict(x=1, s='a'), dict(x=8, s='b'), dict(x=3, s='a')]
      >>> feature_spec = dict(
      ...     x=tf.io.FixedLenFeature([], tf.float32),
      ...     s=tf.io.FixedLenFeature([], tf.string))
      >>> raw_data_metadata = tft.DatasetMetadata.from_feature_spec(feature_spec)
      >>> with tft_beam.Context(temp_dir=tempfile.mkdtemp()):
      ...   transformed_dataset, transform_fn = (
      ...       (raw_data, raw_data_metadata)
      ...       | tft_beam.AnalyzeAndTransformDataset(preprocessing_fn))
      >>> transformed_data, transformed_metadata = transformed_dataset
      >>> transformed_data
      [{'x/mean_a': 0.5}, {'x/mean_a': 4.0}, {'x/mean_a': 1.5}]
  
      Args:
        inputs: An ordered collection of input `Tensor`s.
        ptransform: A Beam PTransform that accepts a Beam PCollection where each
          element is a tuple of `ndarray`s.  Each element in the tuple contains a
          batch of values for the corresponding input tensor of the analyzer and
          maintain their shapes and dtypes.
          It returns a `PCollection`, or a tuple of `PCollections`, each containing
          a single element which is an `ndarray` or a list of primitive types. The
          contents of these output `PCollection`s must be consistent with the given
          values of `output_dtypes` and `output_shapes`.
          It may inherit from `tft_beam.experimental.PTransformAnalyzer` if access
          to a temp base directory is needed.
          Alternatively, it could be an instance of
          `tft.experimental.CacheablePTransformAnalyzer` in order to enable cache
          for this analyzer, when analyzer cache is enabled for this pipeline.
        output_dtypes: An ordered collection of TensorFlow dtypes of the output of
          the analyzer.
        output_shapes: An ordered collection of shapes of the output of the
          analyzer. Must have the same length as output_dtypes.
        output_asset_default_values: (Optional) An ordered collection of optional
          `bytes` aligned with output_dtypes/output_shapes. Every item in this
          collection which is not `None` indicates that the output is a TF asset
          path, and its value would be used as the default value of this asset file
          prior to analysis.
        name: (Optional) Similar to a TF op name.  Used to define a unique scope for
          this analyzer, which can be used for debugging info.
  
      Returns:
        A list of output `Tensor`s.  These will have `dtype` and `shape` as
          specified by `output_dtypes` and `output_shapes`.
  
      Raises:
        ValueError: If output_dtypes and output_shapes have different lengths.
      """
      # pylint: enable=line-too-long
      if len(output_dtypes) != len(output_shapes):
          raise ValueError(
              "output_dtypes ({}) and output_shapes ({}) had different"
              " lengths".format(output_dtypes, output_shapes)
          )
      if output_asset_default_values is not None:
          if len(output_asset_default_values) != len(output_dtypes):
              raise ValueError(
                  "output_dtypes ({}) and output_asset_default_values ({}) had "
                  "different lengths".format(
                      output_dtypes, output_asset_default_values
                  )
              )
          output_asset_default_values = [
              analyzer_nodes.TemporaryAssetInfo(value, "text")
              for value in output_asset_default_values
          ]
      else:
          output_asset_default_values = [None] * len(output_dtypes)
      with tf.compat.v1.name_scope(name, "ptransform"):
          output_tensor_infos = [
              analyzer_nodes.TensorInfo(
                  dtype, shape, default_asset_content
              )
              for dtype, shape, default_asset_content in zip(
                  output_dtypes,
                  output_shapes,
                  output_asset_default_values,
              )
          ]
          return _apply_analyzer(
              ptransform,
              *inputs,
              output_tensor_info_list=output_tensor_infos,
          )
  
imports: from typing import Any, Collection, List, Optional, Tuple, Union, Iterable, Sequence
  import numpy
  import pyarrow
  import tensorflow
  from tensorflow_transform import analyzer_nodes
  from tensorflow_transform import analyzers
  from tensorflow_transform import common
  from tensorflow_transform import common_types
  from tensorflow_transform import nodes
  from tensorflow_transform import tf_utils
  from tfx_bsl import sketches
  from tfx_bsl.types import tfx_namedtuple
  from typing_extensions import Protocol
label_desc: Use image node_id 2 for calling a global function with example usage: ptransform_analyzer(inputs, ptransform, output_dtypes, output_shapes, output_asset_default_values, name) and returns: _apply_analyzer