class_docstr: Kubeflow pipelines task builder.
  
  Constructs a pipeline task spec based on the TFX node information. Meanwhile,
  augments the deployment config associated with the node.
docstr: Builds FileBasedExampleGen into a PipelineContainerSpec.
  
  Returns:
    The PipelineContainerSpec represents the container execution of the
    component, which should includes both the driver execution, and the
    executor execution.
  
  Raises:
    ValueError: When the node is a FileBasedExampleGen but tfx image was not
      specified.
docstr_tok: ["Builds","FileBasedExampleGen","into","a","PipelineContainerSpec",".","Returns",":","The","PipelineContainerSpec","represents","the","container","execution","of","the","component",",","which","should","includes","both","the","driver","execution",",","and","the","executor","execution",".","Raises",":","ValueError",":","When","the","node","is","a","FileBasedExampleGen","but","tfx","image","was","not","specified","."]
code: def _build_file_based_example_gen_spec(self) -> ContainerSpec:
      """Builds FileBasedExampleGen into a PipelineContainerSpec.
  
      Returns:
        The PipelineContainerSpec represents the container execution of the
        component, which should includes both the driver execution, and the
        executor execution.
  
      Raises:
        ValueError: When the node is a FileBasedExampleGen but tfx image was not
          specified.
      """
      assert isinstance(self._node, components.FileBasedExampleGen)
      if not self._tfx_image:
          raise ValueError(
              "TFX image is required for FileBasedExampleGen."
          )
  
      # 1. Build the driver container execution by setting the pre_cache_check
      # hook.
      result = ContainerSpec()
      driver_hook = ContainerSpec.Lifecycle(
          pre_cache_check=ContainerSpec.Lifecycle.Exec(
              command=_DRIVER_COMMANDS,
              args=[
                  "--json_serialized_invocation_args",
                  "{{$}}",
              ],
          )
      )
      driver_hook.pre_cache_check.args.extend(self._beam_pipeline_args)
      result.lifecycle.CopyFrom(driver_hook)
  
      # 2. Build the executor container execution in the same way as a regular
      # component.
      result.image = self._tfx_image
      if self._image_cmds:
          for cmd in self._image_cmds:
              result.command.append(cmd)
      executor_path = name_utils.get_full_name(
          self._node.executor_spec.executor_class
      )
      # Resolve container arguments.
      result.args.append("--executor_class_path")
      result.args.append(executor_path)
      result.args.append("--json_serialized_invocation_args")
      result.args.append("{{$}}")
      result.args.extend(self._beam_pipeline_args)
      return result
  
imports: import itertools
  from typing import Any, Dict, List, Optional, Tuple, cast
  from absl import logging
  from kfp.pipeline_spec import pipeline_spec_pb2
  from tfx import components
  from tfx.components.evaluator import constants
  from tfx.dsl.compiler import compiler_utils
  from tfx.dsl.component.experimental import executor_specs
  from tfx.dsl.component.experimental import placeholders
  from tfx.dsl.components.base import base_component
  from tfx.dsl.components.base import base_node
  from tfx.dsl.components.base import executor_spec
  from tfx.dsl.components.common import importer
  from tfx.dsl.components.common import resolver
  from tfx.dsl.context_managers import dsl_context_registry
  from tfx.dsl.experimental.conditionals import conditional
  from tfx.dsl.input_resolution.strategies import latest_artifact_strategy
  from tfx.dsl.input_resolution.strategies import latest_blessed_model_strategy
  from tfx.dsl.placeholder import placeholder
  from tfx.orchestration import data_types
  from tfx.orchestration.kubeflow import decorators
  from tfx.orchestration.kubeflow import utils
  from tfx.orchestration.kubeflow.v2 import compiler_utils
  from tfx.orchestration.kubeflow.v2 import parameter_utils
  from tfx.types import channel_utils
  from tfx.types import standard_artifacts
  from tfx.types.channel import Channel
  from tfx.utils import deprecation_utils
  from tfx.utils import name_utils
  from ml_metadata.proto import metadata_store_pb2
label_desc: Use image node_id 4 for calling the StepBuilder obj's underlying member method code with example usage: obj._build_file_based_example_gen_spec() and returns: result