class_docstr: Tfx runner on Beam.
class_docstr_tok: ["Tfx","runner","on","Beam","."]
docstr: Deploys given logical pipeline on Beam.
  
  Args:
    tfx_pipeline: Logical pipeline containing pipeline args and components.
docstr_tok: ["Deploys","given","logical","pipeline","on","Beam",".","Args",":","tfx_pipeline",":","Logical","pipeline","containing","pipeline","args","and","components","."]
code: def run(self, tfx_pipeline: pipeline.Pipeline) -> None:
      """Deploys given logical pipeline on Beam.
  
      Args:
        tfx_pipeline: Logical pipeline containing pipeline args and components.
      """
      tfx_pipeline.pipeline_info.run_id = (
          datetime.datetime.now().isoformat()
      )
  
      with telemetry_utils.scoped_labels(
          {telemetry_utils.LABEL_TFX_RUNNER: "beam"}
      ):
          with beam.Pipeline(argv=self._beam_orchestrator_args) as p:
              # Uses for triggering the component DoFns.
              root = p | "CreateRoot" >> beam.Create([None])
  
              # Stores mapping of component to its signal.
              signal_map = {}
              # pipeline.components are in topological order.
              for component in tfx_pipeline.components:
                  # TODO(b/187122662): Pass through pip dependencies as a first-class
                  # component flag.
                  if isinstance(
                      component, base_component.BaseComponent
                  ):
                      component._resolve_pip_dependencies(  # pylint: disable=protected-access
                          tfx_pipeline.pipeline_info.pipeline_root
                      )
                  component_id = component.id
  
                  # Signals from upstream components.
                  signals_to_wait = []
                  if component.upstream_nodes:
                      for upstream_node in component.upstream_nodes:
                          assert upstream_node in signal_map, (
                              "Components is not in "
                              "topological order"
                          )
                          signals_to_wait.append(
                              signal_map[upstream_node]
                          )
                  absl.logging.info(
                      "Component %s depends on %s.",
                      component_id,
                      [s.producer.full_label for s in signals_to_wait],
                  )
  
                  (
                      component_launcher_class,
                      component_config,
                  ) = config_utils.find_component_launch_info(
                      self._config, component
                  )
  
                  # Each signal is an empty PCollection. AsIter ensures component will
                  # be triggered after upstream components are finished.
                  signal_map[
                      component
                  ] = root | "Run[%s]" % component_id >> beam.ParDo(
                      _ComponentAsDoFn(
                          component,
                          component_launcher_class,
                          component_config,
                          tfx_pipeline,
                      ),
                      *[beam.pvalue.AsIter(s) for s in signals_to_wait],
                  )
                  absl.logging.info(
                      "Component %s is scheduled.", component_id
                  )
  
imports: import datetime
  from typing import Any, Iterable, List, Optional, Type
  import absl
  import apache_beam
  from tfx.dsl.components.base import base_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.launcher import base_component_launcher
  from tfx.orchestration.launcher import docker_component_launcher
  from tfx.orchestration.launcher import in_process_component_launcher
  from tfx.utils import telemetry_utils
label_desc: Use image node_id 2 for calling the BeamDagRunner obj's underlying member method code with example usage: obj.run(tfx_pipeline) without return types