class_docstr: TFX bulk inferer executor.
docstr: Runs model inference on given examples data.
  
  Args:
    data_spec: bulk_inferrer_pb2.DataSpec instance.
    output_example_spec: bulk_inferrer_pb2.OutputExampleSpec instance.
    examples: List of `standard_artifacts.Examples` artifacts.
    output_examples: Optional output `standard_artifacts.Examples` artifact.
    inference_result: Optional output `standard_artifacts.InferenceResult`
      artifact.
    inference_endpoint: Model inference endpoint.
docstr_tok: ["Runs","model","inference","on","given","examples","data",".","Args",":","data_spec",":","bulk_inferrer_pb2.DataSpec","instance",".","output_example_spec",":","bulk_inferrer_pb2.OutputExampleSpec","instance",".","examples",":","List","of","`","standard_artifacts.Examples","`","artifacts",".","output_examples",":","Optional","output","`","standard_artifacts.Examples","`","artifact",".","inference_result",":","Optional","output","`","standard_artifacts.InferenceResult","`","artifact",".","inference_endpoint",":","Model","inference","endpoint","."]
code: def _run_model_inference(
      self,
      data_spec: bulk_inferrer_pb2.DataSpec,
      output_example_spec: bulk_inferrer_pb2.OutputExampleSpec,
      examples: List[types.Artifact],
      output_examples: Optional[types.Artifact],
      inference_result: Optional[types.Artifact],
      inference_endpoint: model_spec_pb2.InferenceSpecType,
  ) -> None:
      """Runs model inference on given examples data.
  
      Args:
        data_spec: bulk_inferrer_pb2.DataSpec instance.
        output_example_spec: bulk_inferrer_pb2.OutputExampleSpec instance.
        examples: List of `standard_artifacts.Examples` artifacts.
        output_examples: Optional output `standard_artifacts.Examples` artifact.
        inference_result: Optional output `standard_artifacts.InferenceResult`
          artifact.
        inference_endpoint: Model inference endpoint.
      """
  
      example_uris = {}
      for example_artifact in examples:
          for split in artifact_utils.decode_split_names(
              example_artifact.split_names
          ):
              if data_spec.example_splits:
                  if split in data_spec.example_splits:
                      example_uris[
                          split
                      ] = artifact_utils.get_split_uri(
                          [example_artifact], split
                      )
              else:
                  example_uris[split] = artifact_utils.get_split_uri(
                      [example_artifact], split
                  )
  
      (
          payload_format,
          _,
      ) = tfxio_utils.resolve_payload_format_and_data_view_uri(examples)
  
      tfxio_factory = tfxio_utils.get_tfxio_factory_from_artifact(
          examples,
          _TELEMETRY_DESCRIPTORS,
          schema=None,
          read_as_raw_records=True,
          # We have to specify this parameter in order to create a RawRecord TFXIO
          # but we won't use the RecordBatches so the column name of the raw
          # records does not matter.
          raw_record_column_name="unused",
      )
  
      if output_examples:
          output_examples.split_names = (
              artifact_utils.encode_split_names(
                  sorted(example_uris.keys())
              )
          )
  
      with self._make_beam_pipeline() as pipeline:
          data_list = []
          for split, example_uri in example_uris.items():
              tfxio = tfxio_factory(
                  [io_utils.all_files_pattern(example_uri)]
              )
              assert isinstance(
                  tfxio, record_based_tfxio.RecordBasedTFXIO
              ), "Unable to use TFXIO {} as it does not support reading raw records.".format(
                  type(tfxio)
              )
              # pylint: disable=no-value-for-parameter
              data = (
                  pipeline
                  | "ReadData[{}]".format(split)
                  >> tfxio.RawRecordBeamSource()
                  | "RunInference[{}]".format(split)
                  >> _RunInference(payload_format, inference_endpoint)
              )
              if output_examples:
                  output_examples_split_uri = (
                      artifact_utils.get_split_uri(
                          [output_examples], split
                      )
                  )
                  logging.info(
                      "Path of output examples split `%s` is %s.",
                      split,
                      output_examples_split_uri,
                  )
                  _ = data | "WriteExamples[{}]".format(
                      split
                  ) >> _WriteExamples(
                      output_example_spec, output_examples_split_uri
                  )
                  # pylint: enable=no-value-for-parameter
  
              data_list.append(data)
  
          if inference_result:
              _ = (
                  data_list
                  | "FlattenInferenceResult"
                  >> beam.Flatten(pipeline=pipeline)
                  | "WritePredictionLogs"
                  >> beam.io.WriteToTFRecord(
                      os.path.join(
                          inference_result.uri,
                          _PREDICTION_LOGS_FILE_NAME,
                      ),
                      file_name_suffix=".gz",
                      coder=beam.coders.ProtoCoder(
                          prediction_log_pb2.PredictionLog
                      ),
                  )
              )
  
      if output_examples:
          logging.info(
              "Output examples written to %s.", output_examples.uri
          )
      if inference_result:
          logging.info(
              "Inference result written to %s.", inference_result.uri
          )
  
imports: import importlib
  import os
  from typing import Any, Callable, Dict, List, Optional, Union
  from absl import logging
  import apache_beam
  import tensorflow
  from tfx import types
  from tfx.components.bulk_inferrer import prediction_to_example_utils
  from tfx.components.util import model_utils
  from tfx.components.util import tfxio_utils
  from tfx.dsl.components.base import base_beam_executor
  from tfx.proto import bulk_inferrer_pb2
  from tfx.proto import example_gen_pb2
  from tfx.types import artifact_utils
  from tfx.types import standard_component_specs
  from tfx.utils import io_utils
  from tfx.utils import path_utils
  from tfx.utils import proto_utils
  from tfx_bsl.public.beam import run_inference
  from tfx_bsl.public.proto import model_spec_pb2
  from tfx_bsl.tfxio import record_based_tfxio
  from tensorflow_serving.apis import prediction_log_pb2
label_desc: Use image node_id 3 for calling the Executor obj's underlying member method code with example usage: obj._run_model_inference(data_spec, output_example_spec, examples, output_examples, inference_result, inference_endpoint) without return types