code: def validate_examples_in_tfrecord(
      data_location: Text,
      stats_options: options.StatsOptions,
      output_path: Optional[Text] = None,
      pipeline_options: Optional[PipelineOptions] = None,
      num_sampled_examples=0,
  ) -> Union[
      statistics_pb2.DatasetFeatureStatisticsList,
      Tuple[
          statistics_pb2.DatasetFeatureStatisticsList,
          Mapping[str, List[tf.train.Example]],
      ],
  ]:
      """Validates TFExamples in TFRecord files.
  
      Runs a Beam pipeline to detect anomalies on a per-example basis. If this
      function detects anomalous examples, it generates summary statistics regarding
      the set of examples that exhibit each anomaly.
  
      This is a convenience function for users with data in TFRecord format.
      Users with data in unsupported file/data formats, or users who wish
      to create their own Beam pipelines need to use the 'IdentifyAnomalousExamples'
      PTransform API directly instead.
  
      Args:
        data_location: The location of the input data files.
        stats_options: `tfdv.StatsOptions` for generating data statistics. This must
          contain a schema.
        output_path: The file path to output data statistics result to. If None, the
          function uses a temporary directory. The output will be a TFRecord file
          containing a single data statistics list proto, and can be read with the
          'load_statistics' function.
          If you run this function on Google Cloud, you must specify an
          output_path. Specifying None may cause an error.
        pipeline_options: Optional beam pipeline options. This allows users to
          specify various beam pipeline execution parameters like pipeline runner
          (DirectRunner or DataflowRunner), cloud dataflow service project id, etc.
          See https://cloud.google.com/dataflow/pipelines/specifying-exec-params for
          more details.
        num_sampled_examples: If set, returns up to this many examples
          of each anomaly type as a map from anomaly reason string to a list of
          tf.Examples.
  
      Returns:
        If num_sampled_examples is zero, returns a single
        DatasetFeatureStatisticsList proto in which each dataset consists of the
        set of examples that exhibit a particular anomaly. If
        num_sampled_examples is nonzero, returns the same statistics
        proto as well as a mapping from anomaly to a list of tf.Examples that
        exhibited that anomaly.
  
      Raises:
        ValueError: If the specified stats_options does not include a schema.
      """
      if stats_options.schema is None:
          raise ValueError(
              "The specified stats_options must include a schema."
          )
      if output_path is None:
          output_path = os.path.join(
              tempfile.mkdtemp(), "anomaly_stats.tfrecord"
          )
      output_dir_path = os.path.dirname(output_path)
      if not tf.io.gfile.exists(output_dir_path):
          tf.io.gfile.makedirs(output_dir_path)
      with io_util.Materializer(output_dir_path) as sample_materializer:
          with beam.Pipeline(options=pipeline_options) as p:
              anomalous_examples = (
                  p
                  | "ReadData"
                  >> (
                      tf_example_record.TFExampleRecord(
                          file_pattern=data_location,
                          schema=None,
                          telemetry_descriptors=[
                              "tfdv",
                              "validate_examples_in_tfrecord",
                          ],
                      ).BeamSource(batch_size=1)
                  )
                  | "DetectAnomalies"
                  >> validation_api.IdentifyAnomalousExamples(
                      stats_options
                  )
              )
              _ = (
                  anomalous_examples
                  | "GenerateSummaryStatistics"
                  >> stats_impl.GenerateSlicedStatisticsImpl(
                      stats_options, is_slicing_enabled=True
                  )
                  | "WriteStatsOutput"
                  >> stats_api.WriteStatisticsToTFRecord(output_path)
              )
              if num_sampled_examples:
                  # TODO(b/68154497): Relint
                  # pylint: disable=no-value-for-parameter
                  _ = (
                      anomalous_examples
                      | "Sample"
                      >> beam.combiners.Sample.FixedSizePerKey(
                          num_sampled_examples
                      )
                      | "ToExample"
                      >> _record_batch_to_example_fn(
                          example_coder.RecordBatchToExamplesEncoder(
                              stats_options.schema
                          )
                      )
                      | "WriteSamples" >> sample_materializer.writer()
                  )
                  # pylint: enable=no-value-for-parameter
          if num_sampled_examples:
              samples_per_reason = collections.defaultdict(list)
              for (
                  reason,
                  serialized_example,
              ) in sample_materializer.reader():
                  samples_per_reason[reason].append(
                      tf.train.Example.FromString(serialized_example)
                  )
              return (
                  stats_util.load_statistics(output_path),
                  samples_per_reason,
              )
      return stats_util.load_statistics(output_path)
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  import collections
  import os
  import tempfile
  from typing import List, Mapping, Optional, Text, Tuple, Union
  import apache_beam
  from apache_beam.options.pipeline_options import PipelineOptions
  import pandas
  import pyarrow
  import tensorflow
  from tensorflow_data_validation import types
  from tensorflow_data_validation.api import stats_api
  from tensorflow_data_validation.api import validation_api
  from tensorflow_data_validation.coders import csv_decoder
  from tensorflow_data_validation.statistics import stats_impl
  from tensorflow_data_validation.statistics import stats_options
  from tensorflow_data_validation.utils import io_util
  from tensorflow_data_validation.utils import stats_gen_lib
  from tensorflow_data_validation.utils import stats_util
  from tfx_bsl.coders import example_coder
  from tfx_bsl.tfxio import tf_example_record
  from tensorflow_metadata.proto.v0 import statistics_pb2
label_desc: Use image node_id 3 for calling a global function with example usage: validate_examples_in_tfrecord(data_location, stats_options, output_path, pipeline_options, num_sampled_examples) and returns: stats_util, stats_util,  samples_per_reason