code: def _create_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"]
      )
  
      # 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=_taxi_module_file,
      )
  
      # Uses user-provided Python function that implements a model.
      trainer = Trainer(
          module_file=_taxi_module_file,
          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),
      )
  
      # 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"],
      )
  
      # This custom component serves as a bridge between pipeline and human model
      # reviewers to enable review-and-push workflow in model development cycle. It
      # utilizes Slack API to send message to user-defined Slack channel with model
      # URI info and wait for go / no-go decision from the same Slack channel:
      #   * To approve the model, users need to reply the thread sent out by the bot
      #     started by SlackComponent with 'lgtm' or 'approve'.
      #   * To reject the model, users need to reply the thread sent out by the bot
      #     started by SlackComponent with 'decline' or 'reject'.
      slack_validator = SlackComponent(
          model=trainer.outputs["model"],
          model_blessing=model_validator.outputs["blessing"],
          slack_token=_slack_token,
          slack_channel_id=_slack_channel_id,
          timeout_sec=3600,
      )
  
      # 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=slack_validator.outputs["slack_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,
              evaluator,
              model_validator,
              slack_validator,
              pusher,
          ],
          enable_cache=True,
          metadata_connection_config=metadata.sqlite_metadata_connection_config(
              _metadata_db_root
          ),
      )
  
file_path: tfx/tfx/examples/custom_components/slack/example/taxi_pipeline_slack.py
imports: import datetime
  import os
  from tfx.components import CsvExampleGen
  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.examples.custom_components.slack.slack_component.component import SlackComponent
  from tfx.orchestration import metadata
  from tfx.orchestration import pipeline
  from tfx.orchestration.beam.beam_runner import BeamRunner
  from tfx.proto import evaluator_pb2
  from tfx.proto import pusher_pb2
  from tfx.proto import trainer_pb2
label_desc: Use image node_id 1 for calling a global function with example usage: _create_pipeline() and returns: pipeline