Batch Inference

Path: 3 — ML Engineering on AWS Position: 6 of 12 Difficulty: 🔴 (SageMaker Batch Transform required) Time: 3-4h AWS Cost: ~$0.20–0.50 per batch job (ml.m5.large, 5-10 minutes)

---

Agent Pickup Instructions

# Bootstrap
cd projects/confident-prep/p3-06-batch-inference
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

# Ensure MLflow server is running (from p3-05) and "production" alias exists
# If starting fresh, run p3-05's sweep.py and select_best.py first

# Upload test data and model, run batch transform
python src/prepare_model.py
python src/run_batch.py

# Compare costs
python src/cost_compare.py --batch-duration-minutes 8 --instance-type ml.m5.large \
  --estimated-rpm 100 --endpoint-hourly-cost 0.056

# Run tests
pytest tests/ -v

# TEARDOWN IMMEDIATELY AFTER TESTING
bash scripts/teardown.sh

Done when:

• python src/prepare_model.py creates and uploads model.tar.gz to S3
• python src/run_batch.py completes a SageMaker Batch Transform job and prints accuracy
• Batch output CSV downloaded locally to output/predictions.csv
• python src/cost_compare.py prints batch vs real-time cost comparison
• pytest tests/ -v shows 4/4 passing
• bash scripts/teardown.sh removes S3 inputs, outputs, and SageMaker model

---

What this project is

You take the best model registered in p3-05's MLflow registry ("production" alias), package it as a SageMaker-compatible model.tar.gz, run batch predictions on the held-out test set using SageMaker Batch Transform (a job-based API — no persistent endpoint), download the output CSV, and compare the cost of batch inference versus what a real-time endpoint would cost for the same prediction volume.

What the learner achieves

"I can run SageMaker Batch Transform for offline prediction workloads and explain when batch inference is cheaper than a real-time endpoint — and when it isn't."

---

Folder structure

p3-06-batch-inference/
├── data/
│   ├── adult.data              # full dataset
│   └── test.csv               # held-out test set (features only, no target column)
├── output/
│   └── predictions.csv        # downloaded from S3 after batch job
├── src/
│   ├── prepare_model.py       # load from MLflow, package as tar.gz, upload to S3
│   ├── inference.py           # SageMaker inference handlers (deployed in container)
│   ├── run_batch.py           # create SageMaker Model, run Transformer, download output
│   └── cost_compare.py        # print batch vs real-time cost comparison
├── scripts/
│   └── teardown.sh            # delete SageMaker Model, S3 input + output prefixes
├── tests/
│   ├── test_inference.py      # model_fn loads correctly, output format
│   ├── test_prepare.py        # tar.gz structure validation
│   └── test_cost_compare.py   # known inputs → known outputs
├── .env.example
├── requirements.txt
└── README.md

---

.env.example

# AWS region
AWS_REGION=us-east-1

# S3 bucket for model artifacts and batch data
S3_BUCKET=my-ml-training-bucket

# IAM role ARN with SageMaker and S3 permissions
SAGEMAKER_ROLE_ARN=arn:aws:iam::123456789012:role/SageMakerExecutionRole

# MLflow tracking URI (to load "production" model)
MLFLOW_TRACKING_URI=http://localhost:5000

# MLflow model name matching p3-05 registry
MLFLOW_MODEL_NAME=adult-income-xgboost

# SageMaker model name (must be unique per AWS account/region)
SAGEMAKER_MODEL_NAME=p3-06-adult-income-batch

# XGBoost container image tag (framework version)
XGBOOST_FRAMEWORK_VERSION=1.7-1

---

requirements.txt

boto3==1.35.93
sagemaker==2.236.0
xgboost==2.1.3
mlflow==2.19.0
pandas==2.2.3
numpy==2.2.1
scikit-learn==1.6.0
python-dotenv==1.0.1
pytest==8.3.4

---

src/ — what to implement

src/prepare_model.py

CLI: python src/prepare_model.py [--model-name adult-income-xgboost] [--alias production]

load_model_from_registry(model_name: str, alias: str) -> xgb.XGBClassifier

• Constructs URI f"models:/{model_name}@{alias}"
• Calls mlflow.xgboost.load_model(uri) — returns native XGBClassifier
• If alias not found, raise with message: "No model at alias '{alias}'. Run p3-05 select_best.py first."

create_model_tar(model: xgb.XGBClassifier, output_path: str) -> str

• Saves model to a temp dir as model.xgb using model.save_model()
• Copies src/inference.py into the same temp dir as inference.py
• Creates model.tar.gz containing both files at the root of the archive (not in a subdirectory)
• Writes tar.gz to output_path
• Returns output_path
• Validates: open the tar.gz and assert model.xgb and inference.py are both present

upload_to_s3(local_path: str, bucket: str, key: str) -> str

• Uploads file to s3://{bucket}/{key}
• Returns the S3 URI

prepare_test_data(data_path: str, output_path: str) -> str

• Loads UCI Adult data, applies same preprocessing as training (column names, strip, drop NaN, get_dummies)
• Saves test features only (no target column) to output_path as CSV without index
• Returns output_path

main()

1. Load model from MLflow registry
2. create_model_tar → writes model.tar.gz
3. Upload to s3://{S3_BUCKET}/p3-06/model/model.tar.gz
4. prepare_test_data → writes data/test.csv
5. Upload test CSV to s3://{S3_BUCKET}/p3-06/input/test.csv
6. Print: Model packaged and uploaded. Test data uploaded. Ready for batch transform.

---

src/inference.py

SageMaker inference script — runs inside the XGBoost container. Must handle SageMaker's calling convention.

model_fn(model_dir: str) -> xgb.XGBClassifier

• Loads {model_dir}/model.xgb with xgb.XGBClassifier(); model.load_model(path)
• Returns the model

input_fn(request_body: str, content_type: str) -> np.ndarray

• Accepts content_type == "text/csv"
• Parses CSV string into numpy array using np.frombuffer or pd.read_csv(StringIO(request_body))
• Returns 2D numpy array (float64)
• Raises ValueError(f"Unsupported content type: {content_type}") for other types

predict_fn(input_data: np.ndarray, model: xgb.XGBClassifier) -> np.ndarray

• Calls model.predict_proba(input_data)[:, 1]
• Returns 1D array of probabilities

output_fn(prediction: np.ndarray, accept: str) -> str

• For accept == "text/csv": joins values with newline character
• Returns CSV string

---

src/run_batch.py

CLI: python src/run_batch.py [--instance-type ml.m5.large]

create_sagemaker_model(model_s3_uri: str, role_arn: str, model_name: str, region: str) -> str

• Uses boto3.client('sagemaker') to create a SageMaker Model resource
• Primary container: XGBoost framework container image (fetched via sagemaker.image_uris.retrieve("xgboost", region, version="1.7-1"))
• Environment: {"SAGEMAKER_PROGRAM": "inference.py"}
• Returns model name

run_transform_job(model_name: str, input_s3: str, output_s3: str, instance_type: str) -> str

• Creates and starts a sagemaker.transformer.Transformer with the model
• Calls .transform(input_s3, content_type="text/csv", split_type="Line", wait=True)
• Returns job name

download_and_evaluate(output_s3: str, test_data_path: str, local_output_path: str) -> dict

• Downloads batch output CSV from S3 to local_output_path
• Loads original test labels (re-loads adult.data, extracts y_test at same 80/20 split, random_state=42)
• Converts probability predictions to binary (threshold=0.5)
• Computes accuracy
• Returns {"accuracy": float, "n_predictions": int, "output_path": str}

main()

1. Create SageMaker Model from S3 artifact
2. Run Batch Transform job (wait for completion)
3. Download output and evaluate
4. Print: Batch transform complete. Accuracy: {accuracy:.4f}. Predictions saved to {path}.

---

src/cost_compare.py

CLI: python src/cost_compare.py --batch-duration-minutes N --instance-type T [--estimated-rpm R] [--endpoint-hourly-cost C]

INSTANCE_PRICES: dict[str, float]

INSTANCE_PRICES = {
    "ml.m5.large": 0.115,
    "ml.m5.xlarge": 0.23,
    "ml.t2.medium": 0.056,
}

calculate_batch_cost(instance_type: str, duration_minutes: float) -> float

• Returns INSTANCE_PRICES[instance_type] * (duration_minutes / 60)

calculate_realtime_cost(endpoint_hourly_cost: float, prediction_count: int, rpm: float) -> float

• Time to serve prediction_count predictions at rpm RPM: prediction_count / rpm minutes
• Cost: endpoint_hourly_cost * (minutes / 60)

format_comparison(batch_cost: float, realtime_cost: float, prediction_count: int) -> str

• Returns:

Batch Transform:  ${batch_cost:.4f} for {prediction_count} predictions
Real-time endpoint (estimated): ${realtime_cost:.4f} for same volume
Cheaper option: {'Batch' if batch_cost < realtime_cost else 'Real-time'}
Rule of thumb: Batch is cheaper for infrequent, large-volume jobs. Real-time is
necessary for <1s latency requirements.

---

tests/ — what to test

tests/test_inference.py

test_model_fn_loads_model_correctly

• Create a tiny XGBClassifier, save as model.xgb in a temp dir
• Call inference.model_fn(tmp_dir)
• Assert the returned object is an xgb.XGBClassifier with n_estimators > 0

test_predict_fn_returns_probabilities

• Load tiny model, create 5-row numpy array with correct number of features
• Call inference.predict_fn(arr, model)
• Assert output shape is (5,)
• Assert all values between 0.0 and 1.0

tests/test_prepare.py

test_create_model_tar_has_correct_structure

• Create a tiny XGBClassifier
• Call prepare_model.create_model_tar(model, tmp_path / "model.tar.gz")
• Open the tar.gz and list member names
• Assert "model.xgb" in member names
• Assert "inference.py" in member names

tests/test_cost_compare.py

test_batch_cost_correct_for_known_inputs

• calculate_batch_cost("ml.m5.large", 60) should return 0.115 exactly
• calculate_batch_cost("ml.m5.xlarge", 30) should return 0.115 exactly

test_format_comparison_has_correct_column_count

• Call format_comparison(0.02, 0.15, 10000) — batch cheaper
• Assert output contains "Batch" in the "Cheaper option" line
• Call format_comparison(0.50, 0.02, 100) — real-time cheaper
• Assert output contains "Real-time" in the "Cheaper option" line

---

README.md content

# P3-06 — Batch Inference

SageMaker Batch Transform: offline predictions on held-out test set. No persistent endpoint.

## Prerequisites

- Python 3.11+
- AWS credentials configured
- p3-05 completed: "production" alias in MLflow registry
- MLflow server running at http://localhost:5000 (from p3-05)

## Quick start

```bash
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
cp .env.example .env  # fill in S3_BUCKET and SAGEMAKER_ROLE_ARN

# Package and upload model
python src/prepare_model.py

# Run batch predictions
python src/run_batch.py

# Compare costs
python src/cost_compare.py --batch-duration-minutes 8 --instance-type ml.m5.large \
  --estimated-rpm 100 --endpoint-hourly-cost 0.056

IMPORTANT: teardown after use

bash scripts/teardown.sh

Deletes: SageMaker Model resource, S3 input prefix, S3 output prefix.

Tests

pytest tests/ -v

Tests do not make AWS API calls.

---

## GUIDE.md content

```markdown
# Guide — P3-06 Batch Inference

## Batch Transform vs real-time endpoint

**Batch Transform** is a job: you give it an S3 path of inputs, it spins up instances,
generates predictions, writes outputs to S3, and terminates. You pay only for the duration
of the job. No idle cost.

**Real-time endpoint** is a persistent server: always running, always billable (even when
no requests come in), and returns predictions in <1 second.

**Choose batch when:** predictions can be computed hours in advance (e.g., overnight churn
scores, next-day recommendations). Choose real-time when users are waiting for a response.

## Why model.tar.gz must have files at the root

SageMaker's XGBoost container expects `model.xgb` and `inference.py` at the top level of the
archive (not inside a subdirectory). If you accidentally do `tar czf model.tar.gz mydir/`, the
container will not find your model. Always verify with `tar tzf model.tar.gz` before uploading.

## The split_type="Line" setting

`split_type="Line"` tells SageMaker Batch Transform to send one CSV row per request to
your `input_fn`. Without this, it sends the entire file as one request — which works but
uses more memory and is harder to parallelize. With it, SageMaker can fan out rows across
multiple instances if you scale up.

## Reading batch output

The output file is named `{input_filename}.out` in your S3 output prefix. Each line is
one prediction (the output of your `output_fn`). The ordering matches the input rows.

## Cost math worked out

At ml.m5.large ($0.115/hr), a 10-minute job costs $0.019.
At ml.t2.medium ($0.056/hr) for a real-time endpoint serving 100 RPM for an equivalent
volume of 9,773 predictions: 9,773 / 100 / 60 = 1.6 hours = $0.09.
Batch wins here. Flip the RPM to 10,000 and real-time becomes $0.0009 — batch loses.

## Interview framing

"I use SageMaker Batch Transform for workloads where predictions can be generated offline
in bulk — it's significantly cheaper than a persistent endpoint for infrequent or large-volume
jobs where sub-second latency isn't required."

---

Failure modes to handle

Failure	Where	How to handle
model.xgb and inference.py not at tar root	prepare_model.py: create_model_tar	Always add files with arcname=os.path.basename(file) to keep them at root
MLflow "production" alias not found	prepare_model.py: load_model_from_registry	Raise with message pointing to p3-05
Batch Transform job fails	run_batch.py: run_transform_job	Transformer.wait() raises on failure; catch, print failure reason from describe_transform_job
Output CSV row count doesn't match input	run_batch.py: download_and_evaluate	Log warning; proceed with min(output, input) row count for evaluation
content_type is not CSV	inference.py: input_fn	Raise ValueError with the unsupported type
S3 output prefix already has stale data from prior run	run_batch.py	Batch Transform overwrites; no action needed but document this behavior

---

The metric this project measures

What: Batch prediction accuracy on the 20% holdout test set (9,769 rows)

Format: Accuracy: 0.8715 printed to stdout; n_predictions: 9769

Target: > 0.85 accuracy (same model as p3-05, so performance should match)

---

Cost estimate

Resource	Qty	Rate	Estimated cost
SageMaker Batch Transform (ml.m5.large, ~8 min)	1 job	$0.115/hr	~$0.015
S3 storage: test CSV + output (< 5 MB)	1 session	$0.023/GB-month	<$0.01
S3 requests	~20	$0.005/1000	<$0.01
Total per session			~$0.20–0.50

Teardown checklist

scripts/teardown.sh must:

• Delete SageMaker Model: aws sagemaker delete-model --model-name ${SAGEMAKER_MODEL_NAME}
• Delete S3 input: aws s3 rm s3://${S3_BUCKET}/p3-06/input/ --recursive
• Delete S3 output: aws s3 rm s3://${S3_BUCKET}/p3-06/output/ --recursive
• Delete S3 model artifact: aws s3 rm s3://${S3_BUCKET}/p3-06/model/ --recursive
• Print Teardown complete.