class_docstr: TFX runner on Kubernetes.
class_docstr_tok: ["TFX","runner","on","Kubernetes","."]
docstr: Deploys given logical pipeline on Kubernetes.
  
  Args:
    pipeline: Logical pipeline containing pipeline args and components.
docstr_tok: ["Deploys","given","logical","pipeline","on","Kubernetes",".","Args",":","pipeline",":","Logical","pipeline","containing","pipeline","args","and","components","."]
code: def run(self, pipeline: tfx_pipeline.Pipeline) -> None:
      """Deploys given logical pipeline on Kubernetes.
  
      Args:
        pipeline: Logical pipeline containing pipeline args and components.
      """
      if not pipeline.pipeline_info.run_id:
          pipeline.pipeline_info.run_id = (
              datetime.datetime.now().isoformat()
          )
  
      if not kube_utils.is_inside_cluster():
          kubernetes_remote_runner.run_as_kubernetes_job(
              pipeline=pipeline, tfx_image=self._config.tfx_image
          )
          return
      # TODO(ericlege): Support running components in parallel.
      ran_components = set()
  
      # Runs component in topological order.
      for component in pipeline.components:
          # Verify that components are in topological order.
          if (
              hasattr(component, "upstream_nodes")
              and component.upstream_nodes
          ):
              for upstream_node in component.upstream_nodes:
                  assert upstream_node in ran_components, (
                      "Components is not in " "topological order"
                  )
  
          (
              component_launcher_class,
              component_config,
          ) = config_utils.find_component_launch_info(
              self._config, component
          )
  
          # Check if the component is launchable as a container component.
          if kubernetes_component_launcher.KubernetesComponentLauncher.can_launch(
              component.executor_spec, component_config
          ):
              launch_container_component(
                  component,
                  component_launcher_class,
                  component_config,
                  pipeline,
              )
          # Otherwise, the component should be launchable with the in process
          # component launcher. wrap the component to a container component.
          elif in_process_component_launcher.InProcessComponentLauncher.can_launch(
              component.executor_spec, component_config
          ):
              wrapped_component = self._wrap_container_component(
                  component=component,
                  component_launcher_class=component_launcher_class,
                  component_config=component_config,
                  pipeline=pipeline,
              )
  
              # Component launch info is updated by wrapping the component into a
              # container component. Therefore, these properties need to be reloaded.
              (
                  wrapped_component_launcher_class,
                  wrapped_component_config,
              ) = config_utils.find_component_launch_info(
                  self._config, wrapped_component
              )
  
              launch_container_component(
                  wrapped_component,
                  wrapped_component_launcher_class,
                  wrapped_component_config,
                  pipeline,
              )
          else:
              raise ValueError(
                  "Can not find suitable launcher for component."
              )
  
          ran_components.add(component)
  
imports: import datetime
  import json
  from typing import List, Optional, Type
  from absl import logging
  from tfx.dsl.component.experimental import container_component
  from tfx.dsl.components.base import base_node
  from tfx.orchestration import data_types
  from tfx.orchestration import metadata
  from tfx.orchestration import pipeline
  from tfx.orchestration import tfx_runner
  from tfx.orchestration.config import base_component_config
  from tfx.orchestration.config import config_utils
  from tfx.orchestration.config import pipeline_config
  from tfx.orchestration.experimental.kubernetes import kubernetes_remote_runner
  from tfx.orchestration.experimental.kubernetes import node_wrapper
  from tfx.orchestration.launcher import base_component_launcher
  from tfx.orchestration.launcher import in_process_component_launcher
  from tfx.orchestration.launcher import kubernetes_component_launcher
  from tfx.utils import json_utils
  from tfx.utils import kube_utils
  from tfx.utils import name_utils
  from google.protobuf import json_format
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 2 for calling the KubernetesDagRunner obj's underlying member method code with example usage: obj.run(pipeline) without return types