code: def _MakeTransformOutputAssertFn(self, expected, sort=False):
      # Merge expected instance dicts.
      merged_expected = collections.defaultdict(list)
      for instance_dict in expected:
          for key, value in instance_dict.items():
              # Scalars must be wrapped in a list.
              if (
                  hasattr(value, "__iter__")
                  and not isinstance(value, (str, bytes))
                  or value is None
              ):
                  maybe_wrapped_value = value
              else:
                  maybe_wrapped_value = [value]
              merged_expected[key].append(maybe_wrapped_value)
  
      def _assert_fn(actual):
          # Merge output RecordBatches.
          merged_actual = collections.defaultdict(list)
          for record_batch, _ in actual:
              for key, value in record_batch.to_pydict().items():
                  merged_actual[key].extend(value)
          if sort:
              for value in merged_actual.values():
                  value.sort()
              for value in merged_expected.values():
                  value.sort()
          self.assertDictEqual(merged_expected, merged_actual)
  
      return _assert_fn
  
file_path: transform/tensorflow_transform/beam/impl_output_record_batches_test.py
imports: import collections
  import numpy
  import pyarrow
  import tensorflow
  from tensorflow_transform import impl_helper
  from tensorflow_transform.beam import impl
  from tensorflow_transform.beam import impl_test
  from tensorflow_transform.beam import tft_unit
  from tensorflow_transform.tf_metadata import schema_utils
  from tfx_bsl.tfxio import tensor_adapter
label_desc: Use image node_id 2 for calling the BeamImplOutputRecordBatchesTest obj's underlying member method code with example usage: obj._MakeTransformOutputAssertFn(expected, sort) and returns: _assert_fn