code: def create_pipeline(
      pipeline_name: str,
      pipeline_root: str,
      data_root: str,
      module_file: str,
      ai_platform_training_args: Dict[str, str],
      ai_platform_serving_args: Dict[str, Union[List[str], str]],
      enable_tuning: bool,
      enable_cache: bool,
      user_provided_schema_path: str,
      beam_pipeline_args: List[str],
      use_cloud_component: bool,
      use_aip: bool,
      use_vertex: bool,
      serving_model_dir: Optional[str] = None,
  ) -> tfx.dsl.Pipeline:
      """Implements the penguin pipeline with TFX and Kubeflow Pipeline.
  
      Args:
        pipeline_name: name of the TFX pipeline being created.
        pipeline_root: root directory of the pipeline. Should be a valid GCS path.
        data_root: uri of the penguin data.
        module_file: uri of the module file used in Trainer, Transform and Tuner.
        ai_platform_training_args: Args of CAIP training job. Please refer to
          https://cloud.google.com/ml-engine/reference/rest/v1/projects.jobs#Job for
            detailed description.
        ai_platform_serving_args: Args of CAIP model deployment. Please refer to
          https://cloud.google.com/ml-engine/reference/rest/v1/projects.models for
            detailed description.
        enable_tuning: If True, the hyperparameter tuning through CloudTuner is
          enabled.
        enable_cache: If True, enable caching of pipeline jobs for sequential runs.
        user_provided_schema_path: Path to the schema of the input data.
        beam_pipeline_args: List of Beam pipeline options. Please refer to
          https://cloud.google.com/dataflow/docs/guides/specifying-exec-params#setting-other-cloud-dataflow-pipeline-options.
        use_cloud_component: whether to use tfx.extensions components, namely Tuner,
          Trainer, and Pusher.
        use_aip: whether to use AI platform config with Cloud components; implicitly
          refers to KFP1 orchestration.
        use_vertex: whether to use Vertex config with Cloud components; implicitly
          refers to KFP2 orchestration.
        serving_model_dir: file path to write pipeline saved model to.
  
      Returns:
        A TFX pipeline object.
      """
  
      # Assert Cloud components exist with either AIP/Vertex configuration.
      if use_cloud_component:
          assert (
              use_aip ^ use_vertex
          ), "Cloud component needs either AIP or Vertex configuration."
  
      # TODO(b/248108131): Add an end-to-end test to make sure the runtime parameter
      # is really overriden.
  
      # Number of epochs in training.
      train_args = tfx.dsl.experimental.RuntimeParameter(
          name="train-args",
          default='{"num_steps" : 100}',
          ptype=str,
      )
  
      # Number of epochs in evaluation.
      eval_args = tfx.dsl.experimental.RuntimeParameter(
          name="eval-args",
          default='{"num_steps": 50}',
          ptype=str,
      )
  
      if use_vertex:
          train_args = tfx.proto.TrainArgs(num_steps=100)
          eval_args = tfx.proto.EvalArgs(num_steps=50)
  
      # 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,
      )
  
      # Tunes the hyperparameters for model training based on user-provided Python
      # function. Note that once the hyperparameters are tuned, you can drop the
      # Tuner component from pipeline and feed Trainer with tuned hyperparameters.
      if enable_tuning:
          if use_cloud_component and use_aip:
              # The Tuner component launches 1 AIP Training job for flock management of
              # parallel tuning. For example, 2 workers (defined by num_parallel_trials)
              # in the flock management AIP Training job, each runs a search loop for
              # trials as shown below.
              #   Tuner component -> CAIP job X -> CloudTunerA -> tuning trials
              #                                 -> CloudTunerB -> tuning trials
              #
              # Distributed training for each trial depends on the Tuner
              # (kerastuner.BaseTuner) setup in tuner_fn. Currently CloudTuner is single
              # worker training per trial. DistributingCloudTuner (a subclass of
              # CloudTuner) launches remote distributed training job per trial.
              #
              # E.g., single worker training per trial
              #   ... -> CloudTunerA -> single worker training
              #       -> CloudTunerB -> single worker training
              # vs distributed training per trial
              #   ... -> DistributingCloudTunerA -> CAIP job Y -> master,worker1,2,3
              #       -> DistributingCloudTunerB -> CAIP job Z -> master,worker1,2,3
              tuner = tfx.extensions.google_cloud_ai_platform.Tuner(
                  module_file=module_file,
                  examples=transform.outputs["transformed_examples"],
                  transform_graph=transform.outputs["transform_graph"],
                  train_args=tfx.proto.TrainArgs(num_steps=100),
                  eval_args=tfx.proto.EvalArgs(num_steps=50),
                  tune_args=tfx.proto.TuneArgs(
                      # num_parallel_trials=3 means that 3 search loops are
                      # running in parallel.
                      num_parallel_trials=3
                  ),
                  custom_config={
                      # Note that this TUNING_ARGS_KEY will be used to start the CAIP
                      # job for parallel tuning (CAIP job X above).
                      #
                      # num_parallel_trials will be used to fill/overwrite the
                      # workerCount specified by TUNING_ARGS_KEY:
                      #   num_parallel_trials = workerCount + 1 (for master)
                      tfx.extensions.google_cloud_ai_platform.experimental.TUNING_ARGS_KEY: ai_platform_training_args,
                      # This working directory has to be a valid GCS path and will be
                      # used to launch remote training job per trial.
                      tfx.extensions.google_cloud_ai_platform.experimental.REMOTE_TRIALS_WORKING_DIR_KEY: os.path.join(
                          pipeline_root, "trials"
                      ),
                  },
              )
          elif use_cloud_component and use_vertex:
              tuner = tfx.extensions.google_cloud_ai_platform.Tuner(
                  module_file=module_file,
                  examples=transform.outputs["transformed_examples"],
                  transform_graph=transform.outputs["transform_graph"],
                  train_args=tfx.proto.TrainArgs(num_steps=100),
                  eval_args=tfx.proto.EvalArgs(num_steps=50),
                  tune_args=tfx.proto.TuneArgs(num_parallel_trials=3),
                  custom_config={
                      tfx.extensions.google_cloud_ai_platform.ENABLE_VERTEX_KEY: True,
                      tfx.extensions.google_cloud_ai_platform.VERTEX_REGION_KEY: _gcp_region,
                      tfx.extensions.google_cloud_ai_platform.experimental.TUNING_ARGS_KEY: _vertex_job_spec,
                      tfx.extensions.google_cloud_ai_platform.experimental.REMOTE_TRIALS_WORKING_DIR_KEY: os.path.join(
                          pipeline_root, "trials"
                      ),
                  },
              )
          else:
              tuner = tfx.components.Tuner(
                  examples=transform.outputs["transformed_examples"],
                  transform_graph=transform.outputs["transform_graph"],
                  module_file=module_file,
                  train_args=tfx.proto.TrainArgs(num_steps=100),
                  eval_args=tfx.proto.EvalArgs(num_steps=50),
                  tune_args=tfx.proto.TuneArgs(num_parallel_trials=3),
              )
  
      if use_cloud_component and use_aip:
          # Uses user-provided Python function that trains a model.
          trainer = tfx.extensions.google_cloud_ai_platform.Trainer(
              module_file=module_file,
              examples=transform.outputs["transformed_examples"],
              transform_graph=transform.outputs["transform_graph"],
              schema=schema_gen.outputs["schema"],
              # If Tuner is in the pipeline, Trainer can take Tuner's output
              # best_hyperparameters artifact as input and utilize it in the user
              # module code.
              #
              # If there isn't Tuner in the pipeline, either use Importer to
              # import a previous Tuner's output to feed to Trainer, or directly use
              # the tuned hyperparameters in user module code and set hyperparameters
              # to None here.
              #
              # Example of Importer,
              #   hparams_importer = Importer(
              #     source_uri='path/to/best_hyperparameters.txt',
              #     artifact_type=HyperParameters).with_id('import_hparams')
              #   ...
              #   hyperparameters = hparams_importer.outputs['result'],
              hyperparameters=(
                  tuner.outputs["best_hyperparameters"]
                  if enable_tuning
                  else None
              ),
              train_args=tfx.proto.TrainArgs(num_steps=100),
              eval_args=tfx.proto.EvalArgs(num_steps=50),
              custom_config={
                  tfx.extensions.google_cloud_ai_platform.TRAINING_ARGS_KEY: ai_platform_training_args
              },
          )
      elif use_cloud_component and use_vertex:
          trainer = tfx.extensions.google_cloud_ai_platform.Trainer(
              module_file=module_file,
              examples=transform.outputs["transformed_examples"],
              transform_graph=transform.outputs["transform_graph"],
              schema=schema_gen.outputs["schema"],
              hyperparameters=(
                  tuner.outputs["best_hyperparameters"]
                  if enable_tuning
                  else None
              ),
              train_args=tfx.proto.TrainArgs(num_steps=100),
              eval_args=tfx.proto.EvalArgs(num_steps=50),
              custom_config={
                  tfx.extensions.google_cloud_ai_platform.ENABLE_VERTEX_KEY: True,
                  tfx.extensions.google_cloud_ai_platform.VERTEX_REGION_KEY: _gcp_region,
                  tfx.extensions.google_cloud_ai_platform.TRAINING_ARGS_KEY: _vertex_job_spec,
              },
          )
      else:
          trainer = tfx.components.Trainer(
              module_file=module_file,
              examples=transform.outputs["transformed_examples"],
              transform_graph=transform.outputs["transform_graph"],
              schema=schema_gen.outputs["schema"],
              hyperparameters=(
                  tuner.outputs["best_hyperparameters"]
                  if enable_tuning
                  else None
              ),
              train_args=train_args,
              eval_args=eval_args,
          )
  
      # 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": 0.3}
                              ),
                              # 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,
      )
  
      if use_cloud_component and use_aip:
          pusher = tfx.extensions.google_cloud_ai_platform.Pusher(
              model=trainer.outputs["model"],
              model_blessing=evaluator.outputs["blessing"],
              custom_config={
                  tfx.extensions.google_cloud_ai_platform.experimental.PUSHER_SERVING_ARGS_KEY: ai_platform_serving_args
              },
          )
      elif use_cloud_component and use_vertex:
          pusher = tfx.extensions.google_cloud_ai_platform.Pusher(
              model=trainer.outputs["model"],
              model_blessing=evaluator.outputs["blessing"],
              custom_config={
                  tfx.extensions.google_cloud_ai_platform.ENABLE_VERTEX_KEY: True,
                  tfx.extensions.google_cloud_ai_platform.VERTEX_REGION_KEY: _gcp_region,
                  tfx.extensions.google_cloud_ai_platform.SERVING_ARGS_KEY: _vertex_serving_spec,
              },
          )
      else:
          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
                  )
              ),
          )
  
      components = [
          example_gen,
          statistics_gen,
          schema_gen,
          example_validator,
          transform,
          trainer,
          model_resolver,
          evaluator,
          pusher,
      ]
      if enable_tuning:
          components.append(tuner)
  
      return tfx.dsl.Pipeline(
          pipeline_name=pipeline_name,
          pipeline_root=pipeline_root,
          components=components,
          enable_cache=enable_cache,
          beam_pipeline_args=beam_pipeline_args,
      )
  
file_path: tfx/tfx/examples/penguin/penguin_pipeline_kubeflow.py
imports: import os
  import sys
  from typing import Dict, List, Optional, Union
  from absl import flags
  from absl import logging
  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, ai_platform_training_args, ai_platform_serving_args, enable_tuning, enable_cache, user_provided_schema_path, beam_pipeline_args, use_cloud_component, use_aip, use_vertex, serving_model_dir) and returns: tfx