class_docstr: Responsible for launching a container executor.
class_docstr_tok: ["Responsible","for","launching","a","container","executor","."]
docstr: Execute underlying component implementation.
docstr_tok: ["Execute","underlying","component","implementation","."]
code: def _run_executor(
      self,
      execution_id: int,
      input_dict: Dict[str, List[types.Artifact]],
      output_dict: Dict[str, List[types.Artifact]],
      exec_properties: Dict[str, Any],
  ) -> None:
      """Execute underlying component implementation."""
  
      executor_container_spec = cast(
          executor_spec.ExecutorContainerSpec,
          self._component_executor_spec,
      )
      if self._component_config:
          docker_config = cast(
              docker_component_config.DockerComponentConfig,
              self._component_config,
          )
      else:
          docker_config = (
              docker_component_config.DockerComponentConfig()
          )
  
      # Replace container spec with jinja2 template.
      executor_container_spec = (
          container_common.resolve_container_template(
              executor_container_spec,
              input_dict,
              output_dict,
              exec_properties,
          )
      )
  
      absl.logging.info(
          "Container spec: %s" % vars(executor_container_spec)
      )
      absl.logging.info("Docker config: %s" % vars(docker_config))
  
      # Call client.containers.run and wait for completion.
      # ExecutorContainerSpec follows k8s container spec which has different
      # names to Docker's container spec. It's intended to set command to docker's
      # entrypoint and args to docker's command.
      if docker_config.docker_server_url:
          client = docker.DockerClient(
              base_url=docker_config.docker_server_url
          )
      else:
          client = docker.from_env()
  
      run_args = docker_config.to_run_args()
      container = client.containers.run(
          image=executor_container_spec.image,
          entrypoint=executor_container_spec.command,
          command=executor_container_spec.args,
          detach=True,
          **run_args,
      )
  
      # Streaming logs
      for log in container.logs(stream=True):
          absl.logging.info("Docker: " + log.decode("utf-8"))
      exit_code = container.wait()["StatusCode"]
      if exit_code != 0:
          raise RuntimeError(
              'Container exited with error code "{}"'.format(exit_code)
          )
  
file_path: tfx/tfx/orchestration/launcher/docker_component_launcher.py
imports: from typing import Any, Dict, List, cast
  import absl
  import docker
  from tfx import types
  from tfx.dsl.component.experimental import executor_specs
  from tfx.dsl.components.base import executor_spec
  from tfx.orchestration.config import base_component_config
  from tfx.orchestration.config import docker_component_config
  from tfx.orchestration.launcher import base_component_launcher
  from tfx.orchestration.launcher import container_common
label_desc: Use image node_id 2 for calling the DockerComponentLauncher obj's underlying member method code with example usage: obj._run_executor(execution_id, input_dict, output_dict, exec_properties) without return types