code: def _create_pipeline(
      pipeline_name: str,
      pipeline_root: str,
      data_root: str,
      module_file: str,
      serving_model_dir_lite: str,
      metadata_path: str,
      labels_path: str,
      beam_pipeline_args: List[str],
      accuracy_threshold: float = 0.55,
  ) -> pipeline.Pipeline:
      """Implements the CIFAR10 image classification pipeline using TFX."""
      # This is needed for datasets with pre-defined splits
      # Change the pattern argument to train_whole/* and test_whole/* to train
      # on the whole CIFAR-10 dataset
      input_config = example_gen_pb2.Input(
          splits=[
              example_gen_pb2.Input.Split(
                  name="train", pattern="train/*"
              ),
              example_gen_pb2.Input.Split(
                  name="eval", pattern="test/*"
              ),
          ]
      )
  
      # Brings data into the pipeline.
      example_gen = ImportExampleGen(
          input_base=data_root, input_config=input_config
      )
  
      # Computes statistics over data for visualization and example validation.
      statistics_gen = StatisticsGen(
          examples=example_gen.outputs["examples"]
      )
  
      # Generates schema based on statistics files.
      schema_gen = SchemaGen(
          statistics=statistics_gen.outputs["statistics"],
          infer_feature_shape=True,
      )
  
      # Performs anomaly detection based on statistics and data schema.
      example_validator = ExampleValidator(
          statistics=statistics_gen.outputs["statistics"],
          schema=schema_gen.outputs["schema"],
      )
  
      # Performs transformations and feature engineering in training and serving.
      transform = Transform(
          examples=example_gen.outputs["examples"],
          schema=schema_gen.outputs["schema"],
          module_file=module_file,
      )
  
      # Uses user-provided Python function that trains a model.
      # When traning on the whole dataset, use 18744 for train steps, 156 for eval
      # steps. 18744 train steps correspond to 24 epochs on the whole train set, and
      # 156 eval steps correspond to 1 epoch on the whole test set. The
      # configuration below is for training on the dataset we provided in the data
      # folder, which has 128 train and 128 test samples. The 160 train steps
      # correspond to 40 epochs on this tiny train set, and 4 eval steps correspond
      # to 1 epoch on this tiny test set.
      trainer = Trainer(
          module_file=module_file,
          examples=transform.outputs["transformed_examples"],
          transform_graph=transform.outputs["transform_graph"],
          schema=schema_gen.outputs["schema"],
          train_args=trainer_pb2.TrainArgs(num_steps=160),
          eval_args=trainer_pb2.EvalArgs(num_steps=4),
          custom_config={"labels_path": labels_path},
      )
  
      # Get the latest blessed model for model validation.
      model_resolver = resolver.Resolver(
          strategy_class=latest_blessed_model_resolver.LatestBlessedModelResolver,
          model=Channel(type=Model),
          model_blessing=Channel(type=ModelBlessing),
      ).with_id("latest_blessed_model_resolver")
  
      # Uses TFMA to compute evaluation statistics over features of a model and
      # perform quality validation of a candidate model (compare to a baseline).
      eval_config = tfma.EvalConfig(
          model_specs=[
              tfma.ModelSpec(label_key="label_xf", model_type="tf_lite")
          ],
          slicing_specs=[tfma.SlicingSpec()],
          metrics_specs=[
              tfma.MetricsSpec(
                  metrics=[
                      tfma.MetricConfig(
                          class_name="SparseCategoricalAccuracy",
                          threshold=tfma.MetricThreshold(
                              value_threshold=tfma.GenericValueThreshold(
                                  lower_bound={
                                      "value": accuracy_threshold
                                  }
                              ),
                              # Change threshold will be ignored if there is no
                              # baseline model resolved from MLMD (first run).
                              change_threshold=tfma.GenericChangeThreshold(
                                  direction=tfma.MetricDirection.HIGHER_IS_BETTER,
                                  absolute={"value": -1e-3},
                              ),
                          ),
                      )
                  ]
              )
          ],
      )
  
      # Uses TFMA to compute the evaluation statistics over features of a model.
      # We evaluate using the materialized examples that are output by Transform
      # because
      # 1. the decoding_png function currently performed within Transform are not
      # compatible with TFLite.
      # 2. MLKit requires deserialized (float32) tensor image inputs
      # Note that for deployment, the same logic that is performed within Transform
      # must be reproduced client-side.
      evaluator = Evaluator(
          examples=transform.outputs["transformed_examples"],
          model=trainer.outputs["model"],
          baseline_model=model_resolver.outputs["model"],
          eval_config=eval_config,
      )
  
      # Checks whether the model passed the validation steps and pushes the model
      # to a file destination if check passed.
      pusher = Pusher(
          model=trainer.outputs["model"],
          model_blessing=evaluator.outputs["blessing"],
          push_destination=pusher_pb2.PushDestination(
              filesystem=pusher_pb2.PushDestination.Filesystem(
                  base_directory=serving_model_dir_lite
              )
          ),
      )
  
      components = [
          example_gen,
          statistics_gen,
          schema_gen,
          example_validator,
          transform,
          trainer,
          model_resolver,
          evaluator,
          pusher,
      ]
  
      return pipeline.Pipeline(
          pipeline_name=pipeline_name,
          pipeline_root=pipeline_root,
          components=components,
          enable_cache=True,
          metadata_connection_config=metadata.sqlite_metadata_connection_config(
              metadata_path
          ),
          beam_pipeline_args=beam_pipeline_args,
      )
  
imports: import os
  from typing import List
  import absl
  import tensorflow_model_analysis
  from tfx.components import Evaluator
  from tfx.components import ExampleValidator
  from tfx.components import ImportExampleGen
  from tfx.components import Pusher
  from tfx.components import SchemaGen
  from tfx.components import StatisticsGen
  from tfx.components import Trainer
  from tfx.components import Transform
  from tfx.dsl.components.common import resolver
  from tfx.dsl.experimental import latest_blessed_model_resolver
  from tfx.orchestration import metadata
  from tfx.orchestration import pipeline
  from tfx.orchestration.beam.beam_dag_runner import BeamDagRunner
  from tfx.proto import example_gen_pb2
  from tfx.proto import pusher_pb2
  from tfx.proto import trainer_pb2
  from tfx.types import Channel
  from tfx.types.standard_artifacts import Model
  from tfx.types.standard_artifacts import ModelBlessing
label_desc: Use image node_id 1 for calling a global function with example usage: _create_pipeline(pipeline_name, pipeline_root, data_root, module_file, serving_model_dir_lite, metadata_path, labels_path, beam_pipeline_args, accuracy_threshold) and returns: pipeline