code: def create_pipeline_components(
      pipeline_root: str,
      transform_module: str,
      trainer_module: str,
      bigquery_query: str = "",
      csv_input_location: str = "",
  ) -> List[base_node.BaseNode]:
      """Creates components for a simple Chicago Taxi TFX pipeline for testing.
  
      Args:
        pipeline_root: The root of the pipeline output.
        transform_module: The location of the transform module file.
        trainer_module: The location of the trainer module file.
        bigquery_query: The query to get input data from BigQuery. If not empty,
          BigQueryExampleGen will be used.
        csv_input_location: The location of the input data directory.
  
      Returns:
        A list of TFX components that constitutes an end-to-end test pipeline.
      """
  
      if bool(bigquery_query) == bool(csv_input_location):
          raise ValueError(
              "Exactly one example gen is expected. ",
              "Please provide either bigquery_query or csv_input_location.",
          )
  
      if bigquery_query:
          example_gen = (
              tfx.extensions.google_cloud_big_query.BigQueryExampleGen(
                  query=bigquery_query
              )
          )
      else:
          example_gen = tfx.components.CsvExampleGen(
              input_base=csv_input_location
          )
      statistics_gen = tfx.components.StatisticsGen(
          examples=example_gen.outputs["examples"]
      )
      schema_gen = tfx.components.SchemaGen(
          statistics=statistics_gen.outputs["statistics"],
          infer_feature_shape=False,
      )
      example_validator = tfx.components.ExampleValidator(
          statistics=statistics_gen.outputs["statistics"],
          schema=schema_gen.outputs["schema"],
      )
      transform = tfx.components.Transform(
          examples=example_gen.outputs["examples"],
          schema=schema_gen.outputs["schema"],
          module_file=transform_module,
      )
      latest_model_resolver = tfx.dsl.Resolver(
          strategy_class=tfx.dsl.experimental.LatestArtifactStrategy,
          model=tfx.dsl.Channel(
              type=tfx.types.standard_artifacts.Model
          ),
      ).with_id("Resolver.latest_model_resolver")
      trainer = tfx.components.Trainer(
          custom_executor_spec=executor_spec.ExecutorClassSpec(
              Executor
          ),
          examples=transform.outputs["transformed_examples"],
          schema=schema_gen.outputs["schema"],
          base_model=latest_model_resolver.outputs["model"],
          transform_graph=transform.outputs["transform_graph"],
          train_args=tfx.proto.TrainArgs(num_steps=10),
          eval_args=tfx.proto.EvalArgs(num_steps=5),
          module_file=trainer_module,
      )
      # 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("Resolver.latest_blessed_model_resolver")
      # Set the TFMA config for Model Evaluation and Validation.
      eval_config = tfma.EvalConfig(
          model_specs=[tfma.ModelSpec(signature_name="eval")],
          metrics_specs=[
              tfma.MetricsSpec(
                  metrics=[
                      tfma.MetricConfig(class_name="ExampleCount")
                  ],
                  thresholds={
                      "binary_accuracy": tfma.MetricThreshold(
                          value_threshold=tfma.GenericValueThreshold(
                              lower_bound={"value": 0.5}
                          ),
                          change_threshold=tfma.GenericChangeThreshold(
                              direction=tfma.MetricDirection.HIGHER_IS_BETTER,
                              absolute={"value": -1e-10},
                          ),
                      )
                  },
              )
          ],
          slicing_specs=[
              tfma.SlicingSpec(),
              tfma.SlicingSpec(feature_keys=["trip_start_hour"]),
          ],
      )
      evaluator = tfx.components.Evaluator(
          examples=example_gen.outputs["examples"],
          model=trainer.outputs["model"],
          baseline_model=model_resolver.outputs["model"],
          eval_config=eval_config,
      )
  
      with tfx.dsl.Cond(
          evaluator.outputs["blessing"]
          .future()[0]
          .custom_property("blessed")
          == 1
      ):
          pusher = tfx.components.Pusher(
              model=trainer.outputs["model"],
              push_destination=tfx.proto.PushDestination(
                  filesystem=tfx.proto.PushDestination.Filesystem(
                      base_directory=os.path.join(
                          pipeline_root, "model_serving"
                      )
                  )
              ),
          )
  
      return [
          example_gen,
          statistics_gen,
          schema_gen,
          example_validator,
          transform,
          latest_model_resolver,
          trainer,
          model_resolver,
          evaluator,
          pusher,
      ]
  
imports: import datetime
  import os
  from typing import List
  from kfp.pipeline_spec import pipeline_spec_pb2
  import tensorflow_model_analysis
  from tfx import v1
  from tfx.components.example_gen import utils
  from tfx.components.trainer.executor import Executor
  from tfx.dsl.component.experimental import executor_specs
  from tfx.dsl.component.experimental import placeholders
  from tfx.dsl.components.base import base_component
  from tfx.dsl.components.base import base_executor
  from tfx.dsl.components.base import base_node
  from tfx.dsl.components.base import executor_spec
  from tfx.types import channel_utils
  from tfx.types import component_spec
  from tfx.types.experimental import simple_artifacts
  from tfx.utils import proto_utils
  from google.protobuf import message
label_desc: Use image node_id 6 for calling a global function with example usage: create_pipeline_components(pipeline_root, transform_module, trainer_module, bigquery_query, csv_input_location) and returns: list