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

* 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.

* 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
- Pass S3 URL parameter to script so it can extract filename
- Script constructs local path from filename
- Remove S3 download logic (ProcessingInput handles it)

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.

* 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.

Root cause: ParameterString objects cannot be passed to job_arguments in
SageMaker SDK - they get serialized as '{"Get": "Parameters.Name"}' instead
of being resolved. This is unique to batch_inference template; all other
templates use hardcoded strings.

Solution:
- Use ProcessingInput to download file from S3 URL parameter
- Script finds CSV file in /opt/ml/processing/input/ directory
- Remove --input-data from job_arguments (can't pass parameters there)

This allows users to specify different input data via InputDataUrl parameter
while working around the SDK limitation.

* fix(sagemaker-templates): Convert sparse matrix to dense in batch inference preprocessing

The preprocessing script was outputting sparse matrix format instead of CSV,
causing the Transformer step to fail with 'could not convert string to float'.

Fix: Convert sparse matrix to dense array before writing to CSV.

* style(sagemaker-templates): Fix linting issues in batch inference preprocessing

- Remove unused boto3 import
- Move glob import to top
- Fix line length violations (E501)
- Remove trailing whitespace (W293)

* style(sagemaker-templates): Apply ruff formatting to preprocessing.py

---------

Co-authored-by: Nuri Boardman <nurboard@amazon.com>

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,17 @@ def get_pipeline(
     step_process = ProcessingStep(
         name="PreprocessData",
         processor=sklearn_processor,
+        inputs=[
+            ProcessingInput(
+                source=input_data,
+                destination="/opt/ml/processing/input",
+            ),
+        ],
         outputs=[
             ProcessingOutput(output_name="output_data", source="/opt/ml/processing/output_data"),
         ],
         code="source_scripts/preprocessing.py",
         job_arguments=[
-            "--input-data",
-            input_data,
             "--do-train-test-split",
             "False",
         ],

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

@@ -1,12 +1,12 @@
 """Feature engineers the abalone dataset."""
 
 import argparse
+import glob
 import logging
 import os
 import pathlib
 from typing import Any, Dict
 
-import boto3
 import numpy as np
 import pandas as pd
 from sklearn.compose import ColumnTransformer
@@ -55,23 +55,22 @@ def merge_two_dicts(x: Dict[str, Any], y: Dict[str, Any]) -> Dict[str, Any]:
 if __name__ == "__main__":
     logger.debug("Starting preprocessing.")
     parser = argparse.ArgumentParser()
-    parser.add_argument("--input-data", type=str, required=True)
+    parser.add_argument("--input-data", type=str, required=False)
     parser.add_argument("--do-train-test-split", type=str, default="True")
     args = parser.parse_args()
 
     base_dir = "/opt/ml/processing"
     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)
+    # ProcessingInput downloads file to /opt/ml/processing/input/
+    # Find the CSV file (parameter can't be passed to job_arguments)
+    csv_files = glob.glob("/opt/ml/processing/input/*.csv")
+    if csv_files:
+        fn = csv_files[0]
+        logger.info("Found CSV file: %s", fn)
+    else:
+        raise ValueError("No CSV files found in /opt/ml/processing/input/")
 
-    logger.debug("Reading downloaded data.")
     df = pd.read_csv(
         fn,
         header=None,
@@ -117,4 +116,7 @@ def merge_two_dicts(x: Dict[str, Any], y: Dict[str, Any]) -> Dict[str, Any]:
         pd.DataFrame(test).to_csv(f"{base_dir}/test/test.csv", header=False, index=False)
     else:
         logger.info("Writing out datasets to %s.", base_dir)
+        # Convert sparse matrix to dense array if needed
+        if hasattr(X_pre, "toarray"):
+            X_pre = X_pre.toarray()
         pd.DataFrame(X_pre).to_csv(f"{base_dir}/output_data/data.csv", header=False, index=False)