code: def test_steps_with_map_params_pipeline(
      pipeline_session,
      role,
      script_dir,
      pipeline_name,
      region_name,
      athena_dataset_definition,
  ):
      instance_count = ParameterInteger(
          name="InstanceCount", default_value=2
      )
      framework_version = "0.20.0"
      instance_type = "ml.m5.xlarge"
      output_prefix = ParameterString(
          name="OutputPrefix", default_value="output"
      )
      input_data = f"s3://sagemaker-sample-data-{region_name}/processing/census/census-income.csv"
  
      # The instance_type should not be a pipeline variable
      # since it is used to retrieve image_uri in compile time (PySDK)
      sklearn_processor = SKLearnProcessor(
          framework_version=framework_version,
          instance_type=instance_type,
          instance_count=instance_count,
          base_job_name="test-sklearn",
          sagemaker_session=pipeline_session,
          role=role,
      )
      step_process = ProcessingStep(
          name="my-process",
          display_name="ProcessingStep",
          description="description for Processing step",
          processor=sklearn_processor,
          inputs=[
              ProcessingInput(
                  source=input_data,
                  destination="/opt/ml/processing/input",
              ),
              ProcessingInput(
                  dataset_definition=athena_dataset_definition
              ),
          ],
          outputs=[
              ProcessingOutput(
                  output_name="train_data",
                  source="/opt/ml/processing/train",
              ),
              ProcessingOutput(
                  output_name="test_data",
                  source="/opt/ml/processing/test",
                  destination=Join(
                      on="/",
                      values=[
                          "s3:/",
                          pipeline_session.default_bucket(),
                          "test-sklearn",
                          output_prefix,
                          ExecutionVariables.PIPELINE_EXECUTION_ID,
                      ],
                  ),
              ),
          ],
          code=os.path.join(script_dir, "preprocessing.py"),
      )
  
      # If image_uri is not provided, the instance_type should not be a pipeline variable
      # since instance_type is used to retrieve image_uri in compile time (PySDK)
      sklearn_train = SKLearn(
          framework_version=framework_version,
          entry_point=os.path.join(script_dir, "train.py"),
          instance_type=instance_type,
          sagemaker_session=pipeline_session,
          role=role,
          hyperparameters={
              "batch-size": 500,
              "epochs": 5,
          },
      )
      step_train = TrainingStep(
          name="my-train",
          display_name="TrainingStep",
          description="description for Training step",
          estimator=sklearn_train,
          inputs=TrainingInput(
              s3_data=step_process.properties.ProcessingOutputConfig.Outputs[
                  "train_data"
              ].S3Output.S3Uri
          ),
      )
  
      model = Model(
          image_uri=sklearn_train.image_uri,
          model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
          sagemaker_session=pipeline_session,
          role=role,
      )
      step_model_args = model.create(
          instance_type="ml.m5.large",
          accelerator_type="ml.eia1.medium",
      )
      step_model = ModelStep(
          name="my-model",
          display_name="ModelStep",
          description="description for Model step",
          step_args=step_model_args,
      )
  
      # Condition step for evaluating model quality and branching execution
      cond_lte = ConditionGreaterThanOrEqualTo(
          left=step_train.properties.HyperParameters["batch-size"],
          right=6.0,
      )
  
      step_cond = ConditionStep(
          name="CustomerChurnAccuracyCond",
          conditions=[cond_lte],
          if_steps=[],
          else_steps=[step_model],
      )
  
      pipeline = Pipeline(
          name=pipeline_name,
          parameters=[instance_count, output_prefix],
          steps=[step_process, step_train, step_cond],
          sagemaker_session=pipeline_session,
      )
  
      definition = json.loads(pipeline.definition())
      assert definition["Version"] == "2020-12-01"
  
      steps = definition["Steps"]
      assert len(steps) == 3
      training_args = {}
      condition_args = {}
      for step in steps:
          if step["Type"] == "Training":
              training_args = step["Arguments"]
          if step["Type"] == "Condition":
              condition_args = step["Arguments"]
  
      assert training_args["InputDataConfig"][0]["DataSource"][
          "S3DataSource"
      ]["S3Uri"] == {
          "Get": "Steps.my-process.ProcessingOutputConfig.Outputs['train_data'].S3Output.S3Uri"
      }
      assert condition_args["Conditions"][0]["LeftValue"] == {
          "Get": "Steps.my-train.HyperParameters['batch-size']"
      }
  
      try:
          response = pipeline.create(role)
          create_arn = response["PipelineArn"]
          assert re.match(
              rf"arn:aws:sagemaker:{region_name}:\d{{12}}:pipeline/{pipeline_name}",
              create_arn,
          )
  
      finally:
          try:
              pipeline.delete()
          except Exception:
              pass
  
imports: from __future__ import absolute_import
  import json
  import os
  import re
  import tempfile
  import time
  import shutil
  from contextlib import contextmanager
  import pytest
  import pandas
  from sagemaker.utils import retry_with_backoff
  from tests.integ.sagemaker.workflow.helpers import wait_pipeline_execution
  from tests.integ.s3_utils import extract_files_from_s3
  from sagemaker.workflow.model_step import ModelStep, _REGISTER_MODEL_NAME_BASE, _REPACK_MODEL_NAME_BASE
  from sagemaker.parameter import IntegerParameter
  from sagemaker.pytorch import PyTorch, PyTorchModel
  from sagemaker.tuner import HyperparameterTuner
  from tests.integ.timeout import timeout
  from sagemaker.session import Session
  from sagemaker import image_uris
  from sagemaker.estimator import Estimator
  from sagemaker.inputs import CreateModelInput, TrainingInput
  from sagemaker.model import Model
  from sagemaker.model_metrics import MetricsSource, ModelMetrics
  from sagemaker.processing import ProcessingInput, ProcessingOutput, FeatureStoreOutput, ScriptProcessor
  from sagemaker.s3 import S3Uploader
  from sagemaker.session import get_execution_role
  from sagemaker.sklearn.estimator import SKLearn
  from sagemaker.transformer import Transformer
  from sagemaker.sklearn.processing import SKLearnProcessor
  from sagemaker.workflow.conditions import ConditionGreaterThanOrEqualTo, ConditionLessThanOrEqualTo
  from sagemaker.workflow.condition_step import ConditionStep
  from sagemaker.workflow.callback_step import CallbackStep, CallbackOutput, CallbackOutputTypeEnum
  from sagemaker.wrangler.processing import DataWranglerProcessor
  from sagemaker.dataset_definition.inputs import DatasetDefinition, AthenaDatasetDefinition
  from sagemaker.workflow.execution_variables import ExecutionVariables
  from sagemaker.workflow.functions import Join, JsonGet
  from sagemaker.wrangler.ingestion import generate_data_ingestion_flow_from_s3_input
  from sagemaker.workflow.parameters import ParameterInteger, ParameterString
  from sagemaker.workflow.steps import CreateModelStep, ProcessingStep, TrainingStep, TransformStep, TransformInput, PropertyFile, TuningStep
  from sagemaker.workflow.step_collections import RegisterModel
  from sagemaker.workflow.pipeline import Pipeline
  from sagemaker.workflow.utilities import hash_files_or_dirs
  from sagemaker.feature_store.feature_group import FeatureGroup, FeatureDefinition, FeatureTypeEnum
  from tests.integ import DATA_DIR
label_desc: Use image node_id 9 for calling a global function with example usage: test_steps_with_map_params_pipeline(pipeline_session, role, script_dir, pipeline_name, region_name, athena_dataset_definition) without return types