class_docstr: Abstract TFX executor class for Beam powered components.
class_docstr_tok: ["Abstract","TFX","executor","class","for","Beam","powered","components","."]
docstr: Constructs a beam based executor.
docstr_tok: ["Constructs","a","beam","based","executor","."]
code: def __init__(self, context: Optional[Context] = None):
      """Constructs a beam based executor."""
      super().__init__(context)
  
      self._beam_pipeline_args = []
      self._make_beam_pipeline_fn = None
      if context:
          if isinstance(context, BaseBeamExecutor.Context):
              self._beam_pipeline_args = (
                  context.beam_pipeline_args or []
              )
              self._make_beam_pipeline_fn = (
                  context.make_beam_pipeline_fn
              )
          else:
              raise ValueError(
                  "BaseBeamExecutor found initialized with "
                  "BaseExecutorSpec. Please use BeamExecutorSpec for "
                  "Beam Components instead."
              )
  
      if self._beam_pipeline_args:
          self._beam_pipeline_args = (
              dependency_utils.make_beam_dependency_flags(
                  self._beam_pipeline_args
              )
          )
          executor_class_path = name_utils.get_full_name(self.__class__)
          # TODO(zhitaoli): Rethink how we can add labels and only normalize them
          # if the job is submitted against GCP.
          with telemetry_utils.scoped_labels(
              {telemetry_utils.LABEL_TFX_EXECUTOR: executor_class_path}
          ):
              self._beam_pipeline_args.extend(
                  telemetry_utils.make_beam_labels_args()
              )
  
          # TODO(b/174174381): Don't use beam_pipeline_args to set ABSL flags.
          flags.FLAGS(
              sys.argv + self._beam_pipeline_args, known_only=True
          )
  
file_path: tfx/tfx/dsl/components/base/base_beam_executor.py
imports: import sys
  from typing import Any, Callable, List, Optional, Union
  from absl import flags
  from absl import logging
  from tfx.dsl.components.base.base_executor import BaseExecutor
  from tfx.dsl.placeholder import placeholder
  from tfx.proto.orchestration import pipeline_pb2
  from tfx.utils import name_utils
  from tfx.utils import telemetry_utils
  from tfx.utils import dependency_utils
label_desc: Use image node_id 1 to create a new BaseBeamExecutor object from inherited base classes: BaseExecutor with example: obj = BaseBeamExecutor(context)