code: def testUnbatchExtractor(self):
      model_spec = config_pb2.ModelSpec(
          label_key="label", example_weight_key="example_weight"
      )
      eval_config = config_pb2.EvalConfig(model_specs=[model_spec])
      feature_extractor = features_extractor.FeaturesExtractor(
          eval_config
      )
      label_extractor = labels_extractor.LabelsExtractor(eval_config)
      example_weight_extractor = (
          example_weights_extractor.ExampleWeightsExtractor(eval_config)
      )
      predict_extractor = materialized_predictions_extractor.MaterializedPredictionsExtractor(
          eval_config
      )
      unbatch_inputs_extractor = unbatch_extractor.UnbatchExtractor()
  
      schema = text_format.Parse(
          """
          feature {
            name: "label"
            type: FLOAT
          }
          feature {
            name: "example_weight"
            type: FLOAT
          }
          feature {
            name: "fixed_int"
            type: INT
          }
          feature {
            name: "fixed_float"
            type: FLOAT
          }
          feature {
            name: "fixed_string"
            type: BYTES
          }
          """,
          schema_pb2.Schema(),
      )
      tfx_io = test_util.InMemoryTFExampleRecord(
          schema=schema,
          raw_record_column_name=constants.ARROW_INPUT_COLUMN,
      )
      examples = [
          self._makeExample(
              label=1.0,
              example_weight=0.5,
              fixed_int=1,
              fixed_float=1.0,
              fixed_string="fixed_string1",
          ),
          self._makeExample(
              label=0.0,
              example_weight=0.0,
              fixed_int=1,
              fixed_float=1.0,
              fixed_string="fixed_string2",
          ),
          self._makeExample(
              label=0.0,
              example_weight=1.0,
              fixed_int=2,
              fixed_float=0.0,
              fixed_string="fixed_string3",
          ),
      ]
  
      with beam.Pipeline() as pipeline:
          # pylint: disable=no-value-for-parameter
          result = (
              pipeline
              | "Create"
              >> beam.Create(
                  [e.SerializeToString() for e in examples],
                  reshuffle=False,
              )
              | "BatchExamples" >> tfx_io.BeamSource(batch_size=3)
              | "InputsToExtracts"
              >> model_eval_lib.BatchedInputsToExtracts()
              | feature_extractor.stage_name
              >> feature_extractor.ptransform
              | label_extractor.stage_name >> label_extractor.ptransform
              | example_weight_extractor.stage_name
              >> example_weight_extractor.ptransform
              | predict_extractor.stage_name
              >> predict_extractor.ptransform
              | unbatch_inputs_extractor.stage_name
              >> unbatch_inputs_extractor.ptransform
          )
  
          # pylint: enable=no-value-for-parameter
  
          def check_result(got):
              try:
                  self.assertLen(got, 3)
                  self.assertDictElementsAlmostEqual(
                      got[0][constants.FEATURES_KEY],
                      {
                          "fixed_int": np.array([1]),
                          "fixed_float": np.array([1.0]),
                      },
                  )
                  self.assertEqual(
                      got[0][constants.FEATURES_KEY]["fixed_string"],
                      np.array([b"fixed_string1"]),
                  )
                  self.assertAlmostEqual(
                      got[0][constants.LABELS_KEY], np.array([1.0])
                  )
                  self.assertAlmostEqual(
                      got[0][constants.EXAMPLE_WEIGHTS_KEY],
                      np.array([0.5]),
                  )
                  self.assertDictElementsAlmostEqual(
                      got[1][constants.FEATURES_KEY],
                      {
                          "fixed_int": np.array([1]),
                          "fixed_float": np.array([1.0]),
                      },
                  )
                  self.assertEqual(
                      got[1][constants.FEATURES_KEY]["fixed_string"],
                      np.array([b"fixed_string2"]),
                  )
                  self.assertAlmostEqual(
                      got[1][constants.LABELS_KEY], np.array([0.0])
                  )
                  self.assertAlmostEqual(
                      got[1][constants.EXAMPLE_WEIGHTS_KEY],
                      np.array([0.0]),
                  )
                  self.assertDictElementsAlmostEqual(
                      got[2][constants.FEATURES_KEY],
                      {
                          "fixed_int": np.array([2]),
                          "fixed_float": np.array([0.0]),
                      },
                  )
                  self.assertEqual(
                      got[2][constants.FEATURES_KEY]["fixed_string"],
                      np.array([b"fixed_string3"]),
                  )
                  self.assertAlmostEqual(
                      got[2][constants.LABELS_KEY], np.array([0.0])
                  )
                  self.assertAlmostEqual(
                      got[2][constants.EXAMPLE_WEIGHTS_KEY],
                      np.array([1.0]),
                  )
  
              except AssertionError as err:
                  raise util.BeamAssertException(err)
  
          util.assert_that(result, check_result, label="result")
  
imports: import apache_beam
  from apache_beam.testing import util
  import numpy
  import tensorflow
  from tensorflow_model_analysis import constants
  from tensorflow_model_analysis.api import model_eval_lib
  from tensorflow_model_analysis.eval_saved_model import testutil
  from tensorflow_model_analysis.extractors import example_weights_extractor
  from tensorflow_model_analysis.extractors import features_extractor
  from tensorflow_model_analysis.extractors import labels_extractor
  from tensorflow_model_analysis.extractors import materialized_predictions_extractor
  from tensorflow_model_analysis.extractors import unbatch_extractor
  from tensorflow_model_analysis.proto import config_pb2
  from tfx_bsl.tfxio import test_util
  from google.protobuf import text_format
  from tensorflow_metadata.proto.v0 import schema_pb2
label_desc: Use image node_id 2 for calling the UnbatchExtractorTest obj's underlying member method code with example usage: obj.testUnbatchExtractor() without return types