code: def create_test_pipeline():
      """Builds a parent pipeline that contains multiple inner pipelines."""
      pipeline_name = "composable-pipeline"
      pipeline_root = "pipeline_root"
      serving_model_dir = os.path.join(
          pipeline_root, "serving_model", pipeline_name
      )
      tfx_root = "tfx_root"
      data_path = os.path.join(tfx_root, "data_path")
      pipeline_root = os.path.join(tfx_root, "pipelines", pipeline_name)
  
      data_ingestion = data_ingestion_pipeline(data_path)
  
      training = training_pipeline(
          data_ingestion.outputs["examples"],
          data_ingestion.outputs["schema"].no_trigger(),
      )
  
      eval_config = tfma.EvalConfig(
          model_specs=[tfma.ModelSpec(signature_name="eval")],
          slicing_specs=[tfma.SlicingSpec()],
          metrics_specs=[
              tfma.MetricsSpec(
                  thresholds={
                      "sparse_categorical_accuracy": tfma.MetricThreshold(
                          value_threshold=tfma.GenericValueThreshold(
                              lower_bound={"value": 0.6}
                          ),
                          change_threshold=tfma.GenericChangeThreshold(
                              direction=tfma.MetricDirection.HIGHER_IS_BETTER,
                              absolute={"value": -1e-10},
                          ),
                      )
                  }
              )
          ],
      )
  
      evaluator = Evaluator(
          examples=data_ingestion.outputs["examples"],
          model=training.outputs["model"],
          eval_config=eval_config,
      )
  
      with conditional.Cond(
          evaluator.outputs["blessing"].future()[0].value == 1
      ):
          validate_and_push = validate_and_push_pipeline(
              data_ingestion.outputs["examples"],
              training.outputs["model"],
              evaluator.outputs["blessing"],
              serving_model_dir,
          )
  
      return pipeline.Pipeline(
          pipeline_name=pipeline_name,
          pipeline_root=pipeline_root,
          components=[
              data_ingestion,
              training,
              evaluator,
              validate_and_push,
          ],
          beam_pipeline_args=["--my_testing_beam_pipeline_args=foo"],
          execution_mode=pipeline.ExecutionMode.ASYNC,
      )
  
imports: import os
  import tensorflow_model_analysis
  from tfx.components import CsvExampleGen
  from tfx.components import Evaluator
  from tfx.components import InfraValidator
  from tfx.components import Pusher
  from tfx.components import SchemaGen
  from tfx.components import StatisticsGen
  from tfx.components import Trainer
  from tfx.components.trainer.executor import GenericExecutor
  from tfx.dsl.components.base import executor_spec
  from tfx.dsl.experimental.conditionals import conditional
  from tfx.dsl.experimental.node_execution_options import utils
  from tfx.dsl.placeholder import placeholder
  from tfx.orchestration import pipeline
  from tfx.proto import infra_validator_pb2
  from tfx.proto import pusher_pb2
  from tfx.proto import trainer_pb2
  from tfx.types import channel
label_desc: Use image node_id 5 for calling a global function with example usage: create_test_pipeline() and returns: pipeline