code: def testEvaluateWithQueryBasedMetrics(self):
      temp_export_dir = self._getExportDir()
      (
          _,
          export_dir,
      ) = fixed_prediction_estimator_extra_fields.simple_fixed_prediction_estimator_extra_fields(
          None, temp_export_dir
      )
      eval_config = config_pb2.EvalConfig(
          model_specs=[
              config_pb2.ModelSpec(
                  label_key="label", example_weight_key="fixed_int"
              )
          ],
          slicing_specs=[
              config_pb2.SlicingSpec(),
              config_pb2.SlicingSpec(feature_keys=["fixed_string"]),
          ],
          metrics_specs=metric_specs.specs_from_metrics(
              [ndcg.NDCG(gain_key="fixed_float", name="ndcg")],
              binarize=config_pb2.BinarizationOptions(
                  top_k_list={"values": [1, 2]}
              ),
              query_key="fixed_string",
          ),
      )
      eval_shared_model = self.createTestEvalSharedModel(
          eval_saved_model_path=export_dir,
          tags=[tf.saved_model.SERVING],
      )
      extractors = [
          features_extractor.FeaturesExtractor(eval_config),
          labels_extractor.LabelsExtractor(eval_config),
          example_weights_extractor.ExampleWeightsExtractor(
              eval_config
          ),
          predictions_extractor.PredictionsExtractor(
              eval_shared_model=eval_shared_model,
              eval_config=eval_config,
          ),
          unbatch_extractor.UnbatchExtractor(),
          slice_key_extractor.SliceKeyExtractor(
              eval_config=eval_config
          ),
      ]
      evaluators = [
          metrics_plots_and_validations_evaluator.MetricsPlotsAndValidationsEvaluator(
              eval_config=eval_config,
              eval_shared_model=eval_shared_model,
          )
      ]
  
      # fixed_string used as query_key
      # fixed_float used as gain_key for NDCG
      # fixed_int used as example_weight_key for NDCG
      examples = [
          self._makeExample(
              prediction=0.2,
              label=1.0,
              fixed_float=1.0,
              fixed_string="query1",
              fixed_int=1,
          ),
          self._makeExample(
              prediction=0.8,
              label=0.0,
              fixed_float=0.5,
              fixed_string="query1",
              fixed_int=1,
          ),
          self._makeExample(
              prediction=0.5,
              label=0.0,
              fixed_float=0.5,
              fixed_string="query2",
              fixed_int=2,
          ),
          self._makeExample(
              prediction=0.9,
              label=1.0,
              fixed_float=1.0,
              fixed_string="query2",
              fixed_int=2,
          ),
          self._makeExample(
              prediction=0.1,
              label=0.0,
              fixed_float=0.1,
              fixed_string="query2",
              fixed_int=2,
          ),
          self._makeExample(
              prediction=0.9,
              label=1.0,
              fixed_float=1.0,
              fixed_string="query3",
              fixed_int=3,
          ),
      ]
  
      tfx_io = test_util.InMemoryTFExampleRecord(
          raw_record_column_name=constants.ARROW_INPUT_COLUMN
      )
      with beam.Pipeline() as pipeline:
          # pylint: disable=no-value-for-parameter
          metrics = (
              pipeline
              | "Create"
              >> beam.Create([e.SerializeToString() for e in examples])
              | "BatchExamples" >> tfx_io.BeamSource()
              | "InputsToExtracts"
              >> model_eval_lib.BatchedInputsToExtracts()
              | "ExtractAndEvaluate"
              >> model_eval_lib.ExtractAndEvaluate(
                  extractors=extractors, evaluators=evaluators
              )
          )
  
          # pylint: enable=no-value-for-parameter
  
          def check_metrics(got):
              try:
                  self.assertLen(got, 4)
                  slices = {}
                  for slice_key, value in got:
                      slices[slice_key] = value
                  overall_slice = ()
                  query1_slice = (("fixed_string", "query1"),)
                  query2_slice = (("fixed_string", "query2"),)
                  query3_slice = (("fixed_string", "query3"),)
                  self.assertCountEqual(
                      list(slices),
                      [
                          overall_slice,
                          query1_slice,
                          query2_slice,
                          query3_slice,
                      ],
                  )
                  example_count_key = metric_types.MetricKey(
                      name="example_count"
                  )
                  weighted_example_count_key = metric_types.MetricKey(
                      name="weighted_example_count",
                      example_weighted=True,
                  )
                  ndcg1_key = metric_types.MetricKey(
                      name="ndcg",
                      sub_key=metric_types.SubKey(top_k=1),
                      example_weighted=True,
                  )
                  ndcg2_key = metric_types.MetricKey(
                      name="ndcg",
                      sub_key=metric_types.SubKey(top_k=2),
                      example_weighted=True,
                  )
                  # Query1 (weight=1): (p=0.8, g=0.5) (p=0.2, g=1.0)
                  # Query2 (weight=2): (p=0.9, g=1.0) (p=0.5, g=0.5) (p=0.1, g=0.1)
                  # Query3 (weight=3): (p=0.9, g=1.0)
                  #
                  # DCG@1:  0.5, 1.0, 1.0
                  # NDCG@1: 0.5, 1.0, 1.0
                  # Average NDCG@1: (1 * 0.5 + 2 * 1.0 + 3 * 1.0) / (1 + 2 + 3) ~ 0.92
                  #
                  # DCG@2: (0.5 + 1.0/log(3) ~ 0.630930
                  #        (1.0 + 0.5/log(3) ~ 1.315465
                  #        1.0
                  # NDCG@2: (0.5 + 1.0/log(3)) / (1.0 + 0.5/log(3)) ~ 0.85972
                  #         (1.0 + 0.5/log(3)) / (1.0 + 0.5/log(3)) = 1.0
                  #         1.0
                  # Average NDCG@2: (1 * 0.860 + 2 * 1.0 + 3 * 1.0) / (1 + 2 + 3) ~ 0.97
                  self.assertDictElementsAlmostEqual(
                      slices[overall_slice],
                      {
                          example_count_key: 6,
                          weighted_example_count_key: 11.0,
                          ndcg1_key: 0.9166667,
                          ndcg2_key: 0.9766198,
                      },
                  )
                  self.assertDictElementsAlmostEqual(
                      slices[query1_slice],
                      {
                          example_count_key: 2,
                          weighted_example_count_key: 2.0,
                          ndcg1_key: 0.5,
                          ndcg2_key: 0.85972,
                      },
                  )
                  self.assertDictElementsAlmostEqual(
                      slices[query2_slice],
                      {
                          example_count_key: 3,
                          weighted_example_count_key: 6.0,
                          ndcg1_key: 1.0,
                          ndcg2_key: 1.0,
                      },
                  )
                  self.assertDictElementsAlmostEqual(
                      slices[query3_slice],
                      {
                          example_count_key: 1,
                          weighted_example_count_key: 3.0,
                          ndcg1_key: 1.0,
                          ndcg2_key: 1.0,
                      },
                  )
  
              except AssertionError as err:
                  raise util.BeamAssertException(err)
  
          util.assert_that(
              metrics[constants.METRICS_KEY],
              check_metrics,
              label="metrics",
          )
  
imports: import os
  from absl.testing import parameterized
  import apache_beam
  from apache_beam.testing import util
  import numpy
  import tensorflow
  from tensorflow_model_analysis import constants
  from tensorflow_model_analysis.addons.fairness.metrics import lift
  from tensorflow_model_analysis.api import model_eval_lib
  from tensorflow_model_analysis.eval_saved_model import testutil
  from tensorflow_model_analysis.eval_saved_model.example_trainers import dnn_classifier
  from tensorflow_model_analysis.eval_saved_model.example_trainers import fixed_prediction_estimator_extra_fields
  from tensorflow_model_analysis.eval_saved_model.example_trainers import linear_classifier
  from tensorflow_model_analysis.eval_saved_model.example_trainers import multi_head
  from tensorflow_model_analysis.evaluators import metrics_plots_and_validations_evaluator
  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 legacy_predict_extractor
  from tensorflow_model_analysis.extractors import materialized_predictions_extractor
  from tensorflow_model_analysis.extractors import predictions_extractor
  from tensorflow_model_analysis.extractors import slice_key_extractor
  from tensorflow_model_analysis.extractors import unbatch_extractor
  from tensorflow_model_analysis.metrics import attributions
  from tensorflow_model_analysis.metrics import binary_confusion_matrices
  from tensorflow_model_analysis.metrics import calibration
  from tensorflow_model_analysis.metrics import calibration_plot
  from tensorflow_model_analysis.metrics import confusion_matrix_plot
  from tensorflow_model_analysis.metrics import metric_specs
  from tensorflow_model_analysis.metrics import metric_types
  from tensorflow_model_analysis.metrics import ndcg
  from tensorflow_model_analysis.post_export_metrics import metrics
  from tensorflow_model_analysis.proto import config_pb2
  from tensorflow_model_analysis.proto import validation_result_pb2
  from tfx_bsl.tfxio import raw_tf_record
  from tfx_bsl.tfxio import tensor_adapter
  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 19 for calling the MetricsPlotsAndValidationsEvaluatorTest obj's underlying member method code with example usage: obj.testEvaluateWithQueryBasedMetrics() without return types