code: def _create_pipeline(
      pipeline_name: str,
      pipeline_root: str,
      query: str,
      module_file: str,
      beam_pipeline_args: List[str],
      ai_platform_training_args: Dict[str, str],
      bigquery_serving_args: Dict[str, str],
  ) -> pipeline.Pipeline:
      """Implements the chicago taxi pipeline with TFX and Kubeflow Pipelines."""
  
      # Brings data into the pipeline or otherwise joins/converts training data.
      example_gen = big_query_example_gen_component.BigQueryExampleGen(
          query=query
      )
  
      # 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.
      # to train a model on Google Cloud AI Platform.
      trainer = Trainer(
          custom_executor_spec=executor_spec.ExecutorClassSpec(
              ai_platform_trainer_executor.Executor
          ),
          module_file=module_file,
          transformed_examples=transform.outputs[
              "transformed_examples"
          ],
          schema=schema_gen.outputs["schema"],
          transform_graph=transform.outputs["transform_graph"],
          train_args=trainer_pb2.TrainArgs(num_steps=10000),
          eval_args=trainer_pb2.EvalArgs(num_steps=5000),
          custom_config={
              "ai_platform_training_args": ai_platform_training_args
          },
      )
  
      # Uses TFMA to compute a evaluation statistics over features of a model.
      evaluator = Evaluator(
          examples=example_gen.outputs["examples"],
          model=trainer.outputs["model"],
          feature_slicing_spec=evaluator_pb2.FeatureSlicingSpec(
              specs=[
                  evaluator_pb2.SingleSlicingSpec(
                      column_for_slicing=["trip_start_hour"]
                  )
              ]
          ),
      )
  
      # Performs quality validation of a candidate model (compared to a baseline).
      model_validator = ModelValidator(
          examples=example_gen.outputs["examples"],
          model=trainer.outputs["model"],
      )
  
      # Checks whether the model passed the validation steps and pushes the model
      # to  Google Cloud BigQuery ML if check passed.
      pusher = Pusher(
          custom_executor_spec=executor_spec.ExecutorClassSpec(
              bigquery_pusher_executor.Executor
          ),
          model=trainer.outputs["model"],
          model_blessing=model_validator.outputs["blessing"],
          custom_config={
              "bigquery_serving_args": bigquery_serving_args
          },
      )
  
      return pipeline.Pipeline(
          pipeline_name=pipeline_name,
          pipeline_root=pipeline_root,
          components=[
              example_gen,
              statistics_gen,
              schema_gen,
              example_validator,
              transform,
              trainer,
              evaluator,
              model_validator,
              pusher,
          ],
          beam_pipeline_args=beam_pipeline_args,
      )
  
file_path: tfx/tfx/examples/bigquery_ml/taxi_pipeline_kubeflow_gcp_bqml.py
imports: import os
  from typing import Dict, List
  from tfx.components import Evaluator
  from tfx.components import ExampleValidator
  from tfx.components import ModelValidator
  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.base import executor_spec
  from tfx.extensions.google_cloud_ai_platform.trainer import executor
  from tfx.extensions.google_cloud_big_query.example_gen import component
  from tfx.extensions.google_cloud_big_query.pusher import executor
  from tfx.orchestration import pipeline
  from tfx.orchestration.kubeflow import kubeflow_dag_runner
  from tfx.proto import evaluator_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, query, module_file, beam_pipeline_args, ai_platform_training_args, bigquery_serving_args) and returns: pipeline