code: def _create_pipeline(
      pipeline_name: str,
      pipeline_root: str,
      data_root: str,
      module_file: str,
      accuracy_threshold: float,
      serving_model_dir: str,
      metadata_path: str,
      user_provided_schema_path: str,
      beam_pipeline_args: List[str],
      infra_validator_host_ip_address: str,
      make_warmup: bool,
  ) -> 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"]
      )
  
      # Import user-provided schema.
      schema_gen = tfx.components.ImportSchemaGen(
          schema_file=user_provided_schema_path
      )
  
      # Performs anomaly detection based on statistics and data schema.
      example_validator = tfx.components.ExampleValidator(
          statistics=statistics_gen.outputs["statistics"],
          schema=schema_gen.outputs["schema"],
      )
  
      # Performs transformations and feature engineering in training and serving.
      transform = tfx.components.Transform(
          examples=example_gen.outputs["examples"],
          schema=schema_gen.outputs["schema"],
          module_file=module_file,
      )
  
      # Uses user-provided Python function that trains a model.
      trainer = tfx.components.Trainer(
          module_file=module_file,
          examples=transform.outputs["transformed_examples"],
          transform_graph=transform.outputs["transform_graph"],
          schema=schema_gen.outputs["schema"],
          train_args=tfx.proto.TrainArgs(num_steps=2000),
          eval_args=tfx.proto.EvalArgs(num_steps=5),
      )
  
      # 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(
                  signature_name="serving_default",
                  label_key="species_xf",
                  preprocessing_function_names=["transform_features"],
              )
          ],
          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-10},
                              ),
                          ),
                      )
                  ]
              )
          ],
      )
      evaluator = tfx.components.Evaluator(
          examples=example_gen.outputs["examples"],
          model=trainer.outputs["model"],
          baseline_model=model_resolver.outputs["model"],
          eval_config=eval_config,
      )
  
      # Performs infra validation of a candidate model to prevent unservable model
      # from being pushed. This config will launch a model server of the latest
      # TensorFlow Serving image in a local docker engine.
      infra_validator = tfx.components.InfraValidator(
          model=trainer.outputs["model"],
          examples=example_gen.outputs["examples"],
          serving_spec=tfx.proto.ServingSpec(
              # TODO(b/244254788): Roll back to the 'latest' tag.
              tensorflow_serving=tfx.proto.TensorFlowServing(
                  tags=["2.8.2"]
              ),
              local_docker=tfx.proto.LocalDockerConfig(
                  host_ip_address=infra_validator_host_ip_address
              ),
          ),
          request_spec=tfx.proto.RequestSpec(
              tensorflow_serving=tfx.proto.TensorFlowServingRequestSpec(),
              # If this flag is set, InfraValidator will produce a model with
              # warmup requests (in its outputs['blessing']).
              make_warmup=make_warmup,
          ),
      )
  
      # Checks whether the model passed the validation steps and pushes the model
      # to a file destination if check passed.
      if make_warmup:
          # If InfraValidator.request_spec.make_warmup = True, its output contains
          # a model so that Pusher can push 'infra_blessing' input instead of
          # 'model' input.
          pusher = tfx.components.Pusher(
              model_blessing=evaluator.outputs["blessing"],
              infra_blessing=infra_validator.outputs["blessing"],
              push_destination=tfx.proto.PushDestination(
                  filesystem=tfx.proto.PushDestination.Filesystem(
                      base_directory=serving_model_dir
                  )
              ),
          )
      else:
          # Otherwise, 'infra_blessing' does not contain a model and is used as a
          # conditional checker just like 'model_blessing' does. This is the typical
          # use case.
          pusher = tfx.components.Pusher(
              model=trainer.outputs["model"],
              model_blessing=evaluator.outputs["blessing"],
              infra_blessing=infra_validator.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,
              transform,
              trainer,
              model_resolver,
              evaluator,
              infra_validator,
              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, module_file, accuracy_threshold, serving_model_dir, metadata_path, user_provided_schema_path, beam_pipeline_args, infra_validator_host_ip_address, make_warmup) and returns: tfx