class_docstr: Tfx runner on Beam.
class_docstr_tok: ["Tfx","runner","on","Beam","."]
docstr: Deploys given logical pipeline on Beam.
  
  Args:
    pipeline: Logical pipeline in IR format.
    run_options: Optional args for the run.
  
  Raises:
    ValueError: If run_options is provided, and partial_run_options.from_nodes
      and partial_run_options.to_nodes are both empty.
docstr_tok: ["Deploys","given","logical","pipeline","on","Beam",".","Args",":","pipeline",":","Logical","pipeline","in","IR","format",".","run_options",":","Optional","args","for","the","run",".","Raises",":","ValueError",":","If","run_options","is","provided",",","and","partial_run_options.from_nodes","and","partial_run_options.to_nodes","are","both","empty","."]
code: def run_with_ir(
      self,
      pipeline: pipeline_pb2.Pipeline,
      run_options: Optional[pipeline_pb2.RunOptions] = None,
  ) -> None:
      """Deploys given logical pipeline on Beam.
  
      Args:
        pipeline: Logical pipeline in IR format.
        run_options: Optional args for the run.
  
      Raises:
        ValueError: If run_options is provided, and partial_run_options.from_nodes
          and partial_run_options.to_nodes are both empty.
      """
      run_id = datetime.datetime.now().strftime("%Y%m%d-%H%M%S.%f")
      # Substitute the runtime parameter to be a concrete run_id
      runtime_parameter_utils.substitute_runtime_parameter(
          pipeline,
          {
              constants.PIPELINE_RUN_ID_PARAMETER_NAME: run_id,
          },
      )
  
      deployment_config = self._extract_deployment_config(pipeline)
      connection_config = (
          self._connection_config_from_deployment_config(
              deployment_config
          )
      )
  
      logging.info("Using deployment config:\n %s", deployment_config)
      logging.info("Using connection config:\n %s", connection_config)
  
      if run_options:
          logging.info("Using run_options:\n %s", run_options)
          pr_opts = run_options.partial_run
          partial_run_utils.mark_pipeline(
              pipeline,
              from_nodes=pr_opts.from_nodes or None,
              to_nodes=pr_opts.to_nodes or None,
              snapshot_settings=pr_opts.snapshot_settings,
          )
  
      with telemetry_utils.scoped_labels(
          {telemetry_utils.LABEL_TFX_RUNNER: "beam"}
      ):
          with beam.Pipeline() as p:
              # Uses for triggering the node DoFns.
              root = p | "CreateRoot" >> beam.Create([None])
  
              # Stores mapping of node_id to its signal.
              signal_map = {}
              snapshot_node_id = None
              # pipeline.nodes are in topological order.
              for node in pipeline.nodes:
                  # TODO(b/160882349): Support subpipeline
                  pipeline_node = node.pipeline_node
                  node_id = pipeline_node.node_info.id
                  if pipeline_node.execution_options.HasField("skip"):
                      logging.info(
                          "Node %s is skipped in this partial run.",
                          node_id,
                      )
                      continue
                  run_opts = pipeline_node.execution_options.run
                  if run_opts.perform_snapshot:
                      snapshot_node_id = node_id
                  executor_spec = self._extract_executor_spec(
                      deployment_config, node_id
                  )
                  custom_driver_spec = self._extract_custom_driver_spec(
                      deployment_config, node_id
                  )
  
                  # Signals from upstream nodes.
                  signals_to_wait = []
                  # This ensures that nodes that rely on artifacts reused from previous
                  # pipelines runs are scheduled after the snapshot node.
                  if (
                      run_opts.depends_on_snapshot
                      and node_id != snapshot_node_id
                  ):
                      signals_to_wait.append(
                          signal_map[snapshot_node_id]
                      )
                  for upstream_node_id in pipeline_node.upstream_nodes:
                      if upstream_node_id in signal_map:
                          signals_to_wait.append(
                              signal_map[upstream_node_id]
                          )
                      else:
                          logging.info(
                              "Node %s is upstream of Node %s, but will be skipped in "
                              "this partial run. Node %s is responsible for ensuring that "
                              "Node %s's dependencies can be resolved.",
                              upstream_node_id,
                              node_id,
                              snapshot_node_id,
                              node_id,
                          )
                  logging.info(
                      "Node %s depends on %s.",
                      node_id,
                      [s.producer.full_label for s in signals_to_wait],
                  )
  
                  # Each signal is an empty PCollection. AsIter ensures a node will
                  # be triggered after upstream nodes are finished.
                  signal_map[
                      node_id
                  ] = root | "Run[%s]" % node_id >> beam.ParDo(
                      self._PIPELINE_NODE_DO_FN_CLS(
                          pipeline_node=pipeline_node,
                          mlmd_connection_config=connection_config,
                          pipeline_info=pipeline.pipeline_info,
                          pipeline_runtime_spec=pipeline.runtime_spec,
                          executor_spec=executor_spec,
                          custom_driver_spec=custom_driver_spec,
                          deployment_config=deployment_config,
                          pipeline=pipeline,
                      ),
                      *[beam.pvalue.AsIter(s) for s in signals_to_wait],
                  )
                  logging.info("Node %s is scheduled.", node_id)
  
imports: import datetime
  from typing import Any, Iterable, List, Optional
  from absl import logging
  import apache_beam
  from tfx.dsl.compiler import constants
  from tfx.orchestration import metadata
  from tfx.orchestration.beam.legacy import beam_dag_runner
  from tfx.orchestration.config import pipeline_config
  from tfx.orchestration.local import runner_utils
  from tfx.orchestration.portable import launcher
  from tfx.orchestration.portable import partial_run_utils
  from tfx.orchestration.portable import runtime_parameter_utils
  from tfx.orchestration.portable import tfx_runner
  from tfx.proto.orchestration import local_deployment_config_pb2
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.utils import telemetry_utils
  from google.protobuf import any_pb2
  from google.protobuf import message
label_desc: Use image node_id 8 for calling the BeamDagRunner obj's underlying member method code with example usage: obj.run_with_ir(pipeline, run_options) without return types