code: def _create_pipeline(
      pipeline_name: str,
      pipeline_root: str,
      data_root: str,
      module_file: str,
      serving_model_dir: str,
      metadata_path: str,
      beam_pipeline_args: List[str],
  ) -> pipeline.Pipeline:
      """Implements the chicago taxi pipeline with TFX."""
      # Brings data into the pipeline or otherwise joins/converts training data.
      example_gen = CsvExampleGen(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,
      )
  
      # Uses user-provided Python function that implements a model.
      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=1000),
          eval_args=trainer_pb2.EvalArgs(num_steps=150),
      )
  
      # 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 a evaluation statistics over features of a model and
      # perform quality validation of a candidate model (compared to a baseline).
      eval_config = tfma.EvalConfig(
          model_specs=[
              tfma.ModelSpec(
                  signature_name="serving_default",
                  label_key="tips_xf",
                  preprocessing_function_names=["transform_features"],
              )
          ],
          slicing_specs=[tfma.SlicingSpec()],
          metrics_specs=[
              tfma.MetricsSpec(
                  metrics=[
                      tfma.MetricConfig(
                          class_name="BinaryAccuracy",
                          threshold=tfma.MetricThreshold(
                              value_threshold=tfma.GenericValueThreshold(
                                  lower_bound={"value": 0.6}
                              ),
                              # 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-10},
                              ),
                          ),
                      )
                  ]
              )
          ],
      )
      evaluator = Evaluator(
          examples=example_gen.outputs["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
              )
          ),
      )
  
      return pipeline.Pipeline(
          pipeline_name=pipeline_name,
          pipeline_root=pipeline_root,
          components=[
              example_gen,
              statistics_gen,
              schema_gen,
              example_validator,
              transform,
              trainer,
              model_resolver,
              evaluator,
              pusher,
          ],
          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 CsvExampleGen
  from tfx.components import Evaluator
  from tfx.components import ExampleValidator
  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 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, metadata_path, beam_pipeline_args) and returns: pipeline