class_docstr: TFX bulk inferer executor.
docstr: Runs batch inference on a given model with given input examples.
  
  Args:
    input_dict: Input dict from input key to a list of Artifacts.
      - examples: examples for inference.
      - model: exported model.
      - model_blessing: model blessing result, optional.
    output_dict: Output dict from output key to a list of Artifacts.
      - output: bulk inference results.
    exec_properties: A dict of execution properties.
      - model_spec: JSON string of bulk_inferrer_pb2.ModelSpec instance.
      - data_spec: JSON string of bulk_inferrer_pb2.DataSpec instance.
  
  Returns:
    None
docstr_tok: ["Runs","batch","inference","on","a","given","model","with","given","input","examples",".","Args",":","input_dict",":","Input","dict","from","input","key","to","a","list","of","Artifacts",".","-","examples",":","examples","for","inference",".","-","model",":","exported","model",".","-","model_blessing",":","model","blessing","result",",","optional",".","output_dict",":","Output","dict","from","output","key","to","a","list","of","Artifacts",".","-","output",":","bulk","inference","results",".","exec_properties",":","A","dict","of","execution","properties",".","-","model_spec",":","JSON","string","of","bulk_inferrer_pb2.ModelSpec","instance",".","-","data_spec",":","JSON","string","of","bulk_inferrer_pb2.DataSpec","instance",".","Returns",":","None"]
code: def Do(
      self,
      input_dict: Dict[str, List[types.Artifact]],
      output_dict: Dict[str, List[types.Artifact]],
      exec_properties: Dict[str, Any],
  ) -> None:
      """Runs batch inference on a given model with given input examples.
  
      Args:
        input_dict: Input dict from input key to a list of Artifacts.
          - examples: examples for inference.
          - model: exported model.
          - model_blessing: model blessing result, optional.
        output_dict: Output dict from output key to a list of Artifacts.
          - output: bulk inference results.
        exec_properties: A dict of execution properties.
          - model_spec: JSON string of bulk_inferrer_pb2.ModelSpec instance.
          - data_spec: JSON string of bulk_inferrer_pb2.DataSpec instance.
  
      Returns:
        None
      """
      self._log_startup(input_dict, output_dict, exec_properties)
  
      if output_dict.get(standard_component_specs.INFERENCE_RESULT_KEY):
          inference_result = artifact_utils.get_single_instance(
              output_dict[standard_component_specs.INFERENCE_RESULT_KEY]
          )
      else:
          inference_result = None
      if output_dict.get(standard_component_specs.OUTPUT_EXAMPLES_KEY):
          output_examples = artifact_utils.get_single_instance(
              output_dict[standard_component_specs.OUTPUT_EXAMPLES_KEY]
          )
      else:
          output_examples = None
  
      if "examples" not in input_dict:
          raise ValueError("'examples' is missing in input dict.")
      if "model" not in input_dict:
          raise ValueError(
              "Input models are not valid, model "
              "need to be specified."
          )
      if standard_component_specs.MODEL_BLESSING_KEY in input_dict:
          model_blessing = artifact_utils.get_single_instance(
              input_dict[standard_component_specs.MODEL_BLESSING_KEY]
          )
          if not model_utils.is_model_blessed(model_blessing):
              logging.info(
                  "Model on %s was not blessed", model_blessing.uri
              )
              return
      else:
          logging.info(
              "Model blessing is not provided, exported model will be "
              "used."
          )
  
      model = artifact_utils.get_single_instance(
          input_dict[standard_component_specs.MODEL_KEY]
      )
      model_path = path_utils.serving_model_path(
          model.uri, path_utils.is_old_model_artifact(model)
      )
      logging.info("Use exported model from %s.", model_path)
  
      data_spec = bulk_inferrer_pb2.DataSpec()
      proto_utils.json_to_proto(
          exec_properties[standard_component_specs.DATA_SPEC_KEY],
          data_spec,
      )
  
      output_example_spec = bulk_inferrer_pb2.OutputExampleSpec()
      if exec_properties.get(
          standard_component_specs.OUTPUT_EXAMPLE_SPEC_KEY
      ):
          proto_utils.json_to_proto(
              exec_properties[
                  standard_component_specs.OUTPUT_EXAMPLE_SPEC_KEY
              ],
              output_example_spec,
          )
  
      self._run_model_inference(
          data_spec,
          output_example_spec,
          input_dict[standard_component_specs.EXAMPLES_KEY],
          output_examples,
          inference_result,
          self._get_inference_spec(model_path, exec_properties),
      )
  
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 1 for calling the Executor obj's underlying member method code with example usage: obj.Do(input_dict, output_dict, exec_properties) without return types