code: def _create_pipeline(
      pipeline_name: str,
      pipeline_root: str,
      data_root: str,
      module_file: str,
      module_file_lite: str,
      serving_model_dir: str,
      serving_model_dir_lite: str,
      metadata_path: str,
      beam_pipeline_args: List[str],
      accuracy_threshold: float = 0.8,
  ) -> pipeline.Pipeline:
      """Implements the handwritten digit classification example using TFX."""
      # Brings data into the pipeline.
      example_gen = ImportExampleGen(input_base=data_root)
  
      # 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,
      )
  
      def _create_trainer(module_file, component_id):
          return 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=5000),
              eval_args=trainer_pb2.EvalArgs(num_steps=100),
          ).with_id(component_id)
  
      # Uses user-provided Python function that trains a Keras model.
      trainer = _create_trainer(module_file, "Trainer.mnist")
  
      # Trains the same model as the one above, but converts it into a TFLite one.
      trainer_lite = _create_trainer(
          module_file_lite, "Trainer.mnist_lite"
      )
  
      # TODO(b/150949276): Add resolver back once it supports two trainers.
  
      # Uses TFMA to compute evaluation statistics over features of a model and
      # performs quality validation of a candidate model.
      eval_config = tfma.EvalConfig(
          model_specs=[tfma.ModelSpec(label_key="image_class")],
          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
                                  }
                              )
                          ),
                      )
                  ]
              )
          ],
      )
  
      eval_config_lite = tfma.EvalConfig()
      eval_config_lite.CopyFrom(eval_config)
      # Informs the evaluator that the model is a TFLite model.
      eval_config_lite.model_specs[0].model_type = "tf_lite"
  
      # Uses TFMA to compute the evaluation statistics over features of a model.
      evaluator = Evaluator(
          examples=example_gen.outputs["examples"],
          model=trainer.outputs["model"],
          eval_config=eval_config,
      ).with_id("Evaluator.mnist")
  
      # Uses TFMA to compute the evaluation statistics over features of a TFLite
      # model.
      evaluator_lite = Evaluator(
          examples=example_gen.outputs["examples"],
          model=trainer_lite.outputs["model"],
          eval_config=eval_config_lite,
      ).with_id("Evaluator.mnist_lite")
  
      # 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
              )
          ),
      ).with_id("Pusher.mnist")
  
      # Checks whether the TFLite model passed the validation steps and pushes the
      # model to a file destination if check passed.
      pusher_lite = Pusher(
          model=trainer_lite.outputs["model"],
          model_blessing=evaluator_lite.outputs["blessing"],
          push_destination=pusher_pb2.PushDestination(
              filesystem=pusher_pb2.PushDestination.Filesystem(
                  base_directory=serving_model_dir_lite
              )
          ),
      ).with_id("Pusher.mnist_lite")
  
      return pipeline.Pipeline(
          pipeline_name=pipeline_name,
          pipeline_root=pipeline_root,
          components=[
              example_gen,
              statistics_gen,
              schema_gen,
              example_validator,
              transform,
              trainer,
              trainer_lite,
              evaluator,
              evaluator_lite,
              pusher,
              pusher_lite,
          ],
          enable_cache=True,
          metadata_connection_config=metadata.sqlite_metadata_connection_config(
              metadata_path
          ),
          beam_pipeline_args=beam_pipeline_args,
      )
  
file_path: tfx/tfx/examples/mnist/mnist_pipeline_native_keras.py
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.orchestration import metadata
  from tfx.orchestration import pipeline
  from tfx.orchestration.beam.beam_dag_runner import BeamDagRunner
  from tfx.proto import pusher_pb2
  from tfx.proto import trainer_pb2
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, module_file_lite, serving_model_dir, serving_model_dir_lite, metadata_path, beam_pipeline_args, accuracy_threshold) and returns: pipeline, Trainer