code: def testPreprocessingFn(self):
      schema_file = os.path.join(
          self._testdata_path, "schema_gen/schema.pbtxt"
      )
      schema = io_utils.parse_pbtxt_file(
          schema_file, schema_pb2.Schema()
      )
      feature_spec = taxi_utils._get_raw_feature_spec(schema)
      working_dir = self.get_temp_dir()
      transform_graph_path = os.path.join(
          working_dir, "transform_graph"
      )
      transformed_examples_path = os.path.join(
          working_dir, "transformed_examples"
      )
  
      # Run very simplified version of executor logic.
      # TODO(kestert): Replace with tft_unit.assertAnalyzeAndTransformResults.
      # Generate legacy `DatasetMetadata` object.  Future version of Transform
      # will accept the `Schema` proto directly.
      legacy_metadata = dataset_metadata.DatasetMetadata(
          schema_utils.schema_from_feature_spec(feature_spec)
      )
      tfxio = tf_example_record.TFExampleRecord(
          file_pattern=os.path.join(
              self._testdata_path, "csv_example_gen/Split-train/*"
          ),
          telemetry_descriptors=["Tests"],
          schema=legacy_metadata.schema,
      )
      with beam.Pipeline() as p:
          with tft_beam.Context(
              temp_dir=os.path.join(working_dir, "tmp")
          ):
              examples = p | "ReadTrainData" >> tfxio.BeamSource()
              (
                  transformed_examples,
                  transformed_metadata,
              ), transform_fn = (
                  examples,
                  tfxio.TensorAdapterConfig(),
              ) | "AnalyzeAndTransform" >> tft_beam.AnalyzeAndTransformDataset(
                  taxi_utils.preprocessing_fn
              )
  
              # WriteTransformFn writes transform_fn and metadata to subdirectories
              # tensorflow_transform.SAVED_MODEL_DIR and
              # tensorflow_transform.TRANSFORMED_METADATA_DIR respectively.
              # pylint: disable=expression-not-assigned
              (
                  transform_fn
                  | "WriteTransformFn"
                  >> tft_beam.WriteTransformFn(transform_graph_path)
              )
  
              encoder = tft.coders.ExampleProtoCoder(
                  transformed_metadata.schema
              )
              (
                  transformed_examples
                  | "EncodeTrainData" >> beam.Map(encoder.encode)
                  | "WriteTrainData"
                  >> beam.io.WriteToTFRecord(
                      os.path.join(
                          transformed_examples_path,
                          "Split-train/transformed_examples.gz",
                      ),
                      coder=beam.coders.BytesCoder(),
                  )
              )
              # pylint: enable=expression-not-assigned
  
      # Verify the output matches golden output.
      # NOTE: we don't verify that transformed examples match golden output.
      expected_transformed_schema = io_utils.parse_pbtxt_file(
          os.path.join(
              self._testdata_path,
              "transform/transform_graph/transformed_metadata/schema.pbtxt",
          ),
          schema_pb2.Schema(),
      )
      transformed_schema = io_utils.parse_pbtxt_file(
          os.path.join(
              transform_graph_path, "transformed_metadata/schema.pbtxt"
          ),
          schema_pb2.Schema(),
      )
      # Clear annotations so we only have to test main schema.
      transformed_schema.ClearField("annotation")
      for feature in transformed_schema.feature:
          feature.ClearField("annotation")
      self.assertEqual(transformed_schema, expected_transformed_schema)
  
imports: import os
  import types
  import apache_beam
  import tensorflow
  from tensorflow import estimator
  import tensorflow_model_analysis
  import tensorflow_transform
  from tensorflow_transform import beam
  from tensorflow_transform.tf_metadata import dataset_metadata
  from tensorflow_transform.tf_metadata import schema_utils
  from tfx.components.trainer import executor
  from tfx.components.trainer.fn_args_utils import DataAccessor
  from tfx.components.util import tfxio_utils
  from tfx.dsl.io import fileio
  from tfx.examples.chicago_taxi_pipeline import taxi_utils
  from tfx.types import standard_artifacts
  from tfx.utils import io_utils
  from tfx.utils import path_utils
  from tfx_bsl.tfxio import tf_example_record
  from tensorflow_metadata.proto.v0 import schema_pb2
label_desc: Use image node_id 3 for calling the TaxiUtilsTest obj's underlying member method code with example usage: obj.testPreprocessingFn() without return types