code: def test_stats_impl_custom_generators(self):
      # Dummy PTransform that returns two DatasetFeatureStatistics protos.
      class CustomPTransform(beam.PTransform):
          def expand(self, pcoll):
              stats_proto1 = statistics_pb2.DatasetFeatureStatistics()
              proto1_feat = stats_proto1.features.add()
              proto1_feat.path.step[:] = ["a"]
              custom_stat1 = proto1_feat.custom_stats.add()
              custom_stat1.name = "my_stat_a"
              custom_stat1.str = "my_val_a"
  
              stats_proto2 = statistics_pb2.DatasetFeatureStatistics()
              proto2_feat = stats_proto2.features.add()
              proto2_feat.path.step[:] = ["b"]
              custom_stat2 = proto2_feat.custom_stats.add()
              custom_stat2.name = "my_stat_b"
              custom_stat2.str = "my_val_b"
              return [(None, stats_proto1), (None, stats_proto2)]
  
      record_batches = [
          pa.RecordBatch.from_arrays(
              [
                  pa.array([[]], type=pa.list_(pa.int64())),
                  pa.array([[]], type=pa.list_(pa.int64())),
              ],
              ["a", "b"],
          ),
      ]
      expected_result = text_format.Parse(
          """
      datasets {
        num_examples: 1
        features {
          path {
            step: "a"
          }
          type: INT
          num_stats {
            common_stats {
              num_non_missing: 1
              num_missing: 0
              tot_num_values: 0
            }
          }
          custom_stats {
            name: 'my_stat_a'
            str: 'my_val_a'
          }
        }
        features {
          path {
            step: "b"
          }
          type: INT
          num_stats {
            common_stats {
              num_non_missing: 1
              num_missing: 0
              tot_num_values: 0
            }
          }
          custom_stats {
            name: 'my_stat_b'
            str: 'my_val_b'
          }
        }
      }
      """,
          statistics_pb2.DatasetFeatureStatisticsList(),
      )
  
      # Create a transform stats generator.
      transform_stats_gen = stats_generator.TransformStatsGenerator(
          name="CustomStatsGenerator", ptransform=CustomPTransform()
      )
      with beam.Pipeline() as p:
          options = stats_options.StatsOptions(
              generators=[transform_stats_gen],
              num_values_histogram_buckets=2,
              enable_semantic_domain_stats=True,
          )
          result = (
              p
              | beam.Create(record_batches, reshuffle=False)
              | stats_impl.GenerateStatisticsImpl(options)
          )
          util.assert_that(
              result,
              test_util.make_dataset_feature_stats_list_proto_equal_fn(
                  self, expected_result, check_histograms=False
              ),
          )
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  import copy
  import sys
  from typing import Iterable
  import unittest
  from absl.testing import absltest
  from absl.testing import parameterized
  import apache_beam
  from apache_beam.testing import util
  import numpy
  import pyarrow
  from tensorflow_data_validation.statistics import stats_impl
  from tensorflow_data_validation.statistics import stats_options
  from tensorflow_data_validation.statistics.generators import cross_feature_stats_generator
  from tensorflow_data_validation.statistics.generators import stats_generator
  from tensorflow_data_validation.utils import slicing_util
  from tensorflow_data_validation.utils import test_util
  from tfx_bsl.arrow import array_util
  from tfx_bsl.arrow import table_util
  from tfx_bsl.public.proto import slicing_spec_pb2
  from tfx_bsl.statistics import merge_util
  from google.protobuf import text_format
  from tensorflow.python.util.protobuf import compare
  from tensorflow_metadata.proto.v0 import schema_pb2
  from tensorflow_metadata.proto.v0 import statistics_pb2
label_desc: Use image node_id 7 for calling the StatsImplTest obj's underlying member method code with example usage: obj.test_stats_impl_custom_generators() and returns: list