fix(sagemaker-templates): Fix batch inference parameter resolution

The batch inference template was passing ParameterString objects directly
to job_arguments, which caused them to be serialized as literal JSON strings
instead of being resolved at runtime.

Changes:
- Use ProcessingInput with parameter as source for dynamic S3 paths
- Specify explicit destination filename to avoid path ambiguity
- Follows the same pattern as xgboost_abalone template

This fixes the issue where InputDataUrl parameter was passed as
'{"Get": "Parameters.InputDataUrl"}' instead of the actual S3 path,
causing preprocessing jobs to fail with IndexError.

Resolves parameter resolution for batch inference pipelines.

Author: newman911

modules/sagemaker/sagemaker-templates/templates/batch_inference/seed_code/build_app/ml_pipelines/transformer/pipeline.py

@@ -6,7 +6,7 @@
 import sagemaker.session
 from sagemaker import ModelPackage
 from sagemaker.inputs import TransformInput
-from sagemaker.processing import ProcessingOutput
+from sagemaker.processing import ProcessingInput, ProcessingOutput
 from sagemaker.sklearn.processing import SKLearnProcessor
 from sagemaker.transformer import Transformer
 from sagemaker.workflow.execution_variables import ExecutionVariables
@@ -141,13 +141,19 @@ def get_pipeline(
     step_process = ProcessingStep(
         name="PreprocessData",
         processor=sklearn_processor,
+        inputs=[
+            ProcessingInput(
+                source=input_data,
+                destination="/opt/ml/processing/input/data.csv",
+            ),
+        ],
         outputs=[
             ProcessingOutput(output_name="output_data", source="/opt/ml/processing/output_data"),
         ],
         code="source_scripts/preprocessing.py",
         job_arguments=[
             "--input-data",
-            input_data,
+            "/opt/ml/processing/input/data.csv",
             "--do-train-test-split",
             "False",
         ],

modules/sagemaker/sagemaker-templates/templates/batch_inference/seed_code/build_app/source_scripts/preprocessing.py

@@ -63,13 +63,20 @@ def merge_two_dicts(x: Dict[str, Any], y: Dict[str, Any]) -> Dict[str, Any]:
     pathlib.Path(f"{base_dir}/data").mkdir(parents=True, exist_ok=True)
     input_data = args.input_data
     logger.info("Input data path: %s", input_data)
-    bucket = input_data.split("/")[2]
-    key = "/".join(input_data.split("/")[3:])
-
-    logger.info("Downloading data from bucket: %s, key: %s", bucket, key)
-    fn = f"{base_dir}/data/abalone-dataset.csv"
-    s3 = boto3.resource("s3")
-    s3.Bucket(bucket).download_file(key, fn)
+    
+    # Check if input_data is an S3 URL or a local path
+    if input_data.startswith("s3://"):
+        # Download from S3
+        bucket = input_data.split("/")[2]
+        key = "/".join(input_data.split("/")[3:])
+        logger.info("Downloading data from bucket: %s, key: %s", bucket, key)
+        fn = f"{base_dir}/data/abalone-dataset.csv"
+        s3 = boto3.resource("s3")
+        s3.Bucket(bucket).download_file(key, fn)
+    else:
+        # Use local path (file already downloaded by ProcessingInput)
+        logger.info("Using local input data path")
+        fn = input_data
 
     logger.debug("Reading downloaded data.")
     df = pd.read_csv(