code: def _create_pipeline(
      pipeline_name: str,
      pipeline_root: str,
      data_root: str,
      trainer_module_file: str,
      evaluator_module_file: str,
      serving_model_dir: str,
      metadata_path: str,
      beam_pipeline_args: List[str],
  ) -> tfx.dsl.Pipeline:
      """Implements the Penguin pipeline with TFX."""
      # Brings data into the pipeline or otherwise joins/converts training data.
      example_gen = tfx.components.CsvExampleGen(
          input_base=os.path.join(data_root, "labelled")
      )
  
      # Computes statistics over data for visualization and example validation.
      statistics_gen = tfx.components.StatisticsGen(
          examples=example_gen.outputs["examples"]
      )
  
      # Generates schema based on statistics files.
      schema_gen = tfx.components.SchemaGen(
          statistics=statistics_gen.outputs["statistics"],
          infer_feature_shape=True,
      )
  
      # Performs anomaly detection based on statistics and data schema.
      example_validator = tfx.components.ExampleValidator(
          statistics=statistics_gen.outputs["statistics"],
          schema=schema_gen.outputs["schema"],
      )
  
      # TODO(humichael): Handle applying transformation component in Milestone 3.
  
      # Uses user-provided Python function that trains a model.
      # Num_steps is not provided during evaluation because the scikit-learn model
      # loads and evaluates the entire test set at once.
      trainer = tfx.components.Trainer(
          module_file=trainer_module_file,
          examples=example_gen.outputs["examples"],
          schema=schema_gen.outputs["schema"],
          train_args=tfx.proto.TrainArgs(num_steps=2000),
          eval_args=tfx.proto.EvalArgs(),
      )
  
      # Get the latest blessed model for model validation.
      model_resolver = tfx.dsl.Resolver(
          strategy_class=tfx.dsl.experimental.LatestBlessedModelStrategy,
          model=tfx.dsl.Channel(
              type=tfx.types.standard_artifacts.Model
          ),
          model_blessing=tfx.dsl.Channel(
              type=tfx.types.standard_artifacts.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 (compared to a baseline).
      eval_config = tfma.EvalConfig(
          model_specs=[tfma.ModelSpec(label_key="species")],
          slicing_specs=[tfma.SlicingSpec()],
          metrics_specs=[
              tfma.MetricsSpec(
                  metrics=[
                      tfma.MetricConfig(
                          class_name="Accuracy",
                          threshold=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 = tfx.components.Evaluator(
          module_file=evaluator_module_file,
          examples=example_gen.outputs["examples"],
          model=trainer.outputs["model"],
          baseline_model=model_resolver.outputs["model"],
          eval_config=eval_config,
      )
  
      pusher = tfx.components.Pusher(
          model=trainer.outputs["model"],
          model_blessing=evaluator.outputs["blessing"],
          push_destination=tfx.proto.PushDestination(
              filesystem=tfx.proto.PushDestination.Filesystem(
                  base_directory=serving_model_dir
              )
          ),
      )
  
      return tfx.dsl.Pipeline(
          pipeline_name=pipeline_name,
          pipeline_root=pipeline_root,
          components=[
              example_gen,
              statistics_gen,
              schema_gen,
              example_validator,
              trainer,
              model_resolver,
              evaluator,
              pusher,
          ],
          enable_cache=True,
          metadata_connection_config=tfx.orchestration.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 import v1
label_desc: Use image node_id 1 for calling a global function with example usage: _create_pipeline(pipeline_name, pipeline_root, data_root, trainer_module_file, evaluator_module_file, serving_model_dir, metadata_path, beam_pipeline_args) and returns: tfx