Feature Engineering

Path: 3 — ML Engineering on AWS Position: 3 of 12 Difficulty: 🟡 (local pipeline works without AWS; SageMaker Processing is the scale step) Time: 4-5h local, +1h with SageMaker Processing AWS Cost: ~$0.10 per Processing Job (ml.m5.large, ~5 minutes)

---

Agent Pickup Instructions

# Bootstrap
cd projects/confident-prep/p3-03-feature-engineering
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

# Download dataset (same as p3-01 and p3-02)
mkdir -p data
curl -o data/raw.csv \
  "https://archive.ics.uci.edu/ml/machine-learning-databases/adult/adult.data"

# Run locally
python src/main.py --input data/raw.csv --output data/processed.csv --save-pipeline pipeline.joblib

# Run tests
pytest tests/ -v

# Submit to SageMaker Processing (requires AWS env vars)
python src/sagemaker_processor.py

Done when:

• python src/main.py --input data/raw.csv --output data/processed.csv exits 0 and writes data/processed.csv
• data/processed.csv has no NaN values (verify with python -c "import pandas as pd; assert not pd.read_csv('data/processed.csv').isna().any().any()")
• pipeline.joblib is written when --save-pipeline is passed
• Loaded pipeline can transform new data without re-fitting
• pytest tests/ -v shows 4/4 passing
• bash scripts/teardown.sh removes S3 Processing output prefix

---

What this project is

A reusable sklearn preprocessing Pipeline for the UCI Adult Income dataset. The pipeline applies median imputation to numeric columns, most-frequent imputation to categorical columns, standard scaling to numeric columns, and one-hot encoding to categorical columns — all composed via a ColumnTransformer. Data validation runs before AND after transformation. The fitted pipeline is serialized with joblib. The same main.py runs locally and as a SageMaker Processing Job.

What the learner achieves

"I can build a reproducible sklearn preprocessing pipeline, validate its inputs and outputs, serialize it, and run it identically on a local machine and in SageMaker Processing — so preprocessing is never ad-hoc."

---

Folder structure

p3-03-feature-engineering/
├── data/
│   ├── raw.csv                 # UCI Adult data (renamed from adult.data)
│   └── processed.csv          # output of pipeline
├── src/
│   ├── main.py                # CLI: --input, --output, [--save-pipeline]
│   ├── pipeline.py            # build_pipeline, fit_and_transform, transform
│   ├── validator.py           # validate_input (schema), validate_output (no NaN, shape)
│   └── sagemaker_processor.py # submit main.py to SageMaker SKLearnProcessor
├── scripts/
│   └── teardown.sh            # delete S3 Processing Job output prefix
├── tests/
│   ├── test_pipeline.py       # no NaN output, joblib roundtrip, shape non-zero
│   └── test_validator.py      # missing column raises ValueError, NaN output raises ValueError
├── .env.example
├── requirements.txt
└── README.md

---

.env.example

# AWS region (needed for SageMaker Processing only)
AWS_REGION=us-east-1

# S3 bucket for Processing Job input and output
S3_BUCKET=my-ml-training-bucket

# IAM role ARN that SageMaker will assume
SAGEMAKER_ROLE_ARN=arn:aws:iam::123456789012:role/SageMakerExecutionRole

# Local paths (no AWS required)
LOCAL_DATA_DIR=data/
LOCAL_PIPELINE_PATH=pipeline.joblib

---

requirements.txt

scikit-learn==1.6.0
pandas==2.2.3
numpy==2.2.1
joblib==1.4.2
boto3==1.35.93
sagemaker==2.236.0
python-dotenv==1.0.1
pytest==8.3.4

---

src/ — what to implement

src/pipeline.py

Column lists (module-level constants):

NUMERIC_FEATURES = ["age", "fnlwgt", "education-num",
                    "capital-gain", "capital-loss", "hours-per-week"]
CATEGORICAL_FEATURES = ["workclass", "education", "marital-status",
                        "occupation", "relationship", "race",
                        "sex", "native-country"]
TARGET_COLUMN = "income"

build_pipeline() -> sklearn.pipeline.Pipeline

• Creates a ColumnTransformer with two branches:
  • numeric: Pipeline([("imputer", SimpleImputer(strategy="median")), ("scaler", StandardScaler())])
  • categorical: Pipeline([("imputer", SimpleImputer(strategy="most_frequent")), ("encoder", OneHotEncoder(handle_unknown="ignore", sparse_output=False))])
• Wraps ColumnTransformer in a top-level Pipeline (no estimator — preprocessing only)
• Returns the unfitted pipeline

fit_and_transform(pipeline: Pipeline, X: pd.DataFrame) -> np.ndarray

• Calls pipeline.fit_transform(X)
• Returns the resulting numpy array
• Does NOT modify pipeline in place beyond what fit_transform does

transform(pipeline: Pipeline, X: pd.DataFrame) -> np.ndarray

• Calls pipeline.transform(X) — pipeline must already be fitted
• Raises sklearn.exceptions.NotFittedError (propagated naturally) if called before fitting

get_feature_names(pipeline: Pipeline) -> list[str]

• Extracts output feature names from the fitted ColumnTransformer
• Returns a flat list of strings: ["age", "fnlwgt", ..., "workclass_Federal-gov", ...]
• Uses pipeline.named_steps['preprocessor'].get_feature_names_out()

---

src/validator.py

REQUIRED_COLUMNS: list[str] — same column names as p3-01/p3-02 (all 14 feature columns, excluding income)

validate_input(df: pd.DataFrame, expected_columns: list[str] | None = None) -> None

• If expected_columns is None, uses REQUIRED_COLUMNS
• Checks that every column in expected_columns appears in df.columns
• Raises ValueError(f"Missing columns: {missing_cols}") if any are absent
• Checks that df has at least 1 row; raises ValueError("DataFrame is empty") if not
• Does NOT modify the DataFrame

validate_output(arr: np.ndarray) -> None

• Checks arr.ndim == 2; raises ValueError("Output must be 2D array") if not
• Checks arr.shape[0] > 0; raises ValueError("Output has zero rows") if not
• Checks not np.isnan(arr).any(); raises ValueError("Output contains NaN values after transformation") if NaN found
• Does NOT check dtype (float32 and float64 are both valid)

---

src/main.py

CLI arguments:

• --input (str, required) — path to raw CSV
• --output (str, required) — path to write processed CSV
• --save-pipeline (str, optional) — path to save fitted joblib pipeline

load_raw(input_path: str) -> tuple[pd.DataFrame, pd.Series]

• Re-uses loader logic from p3-02 (or copies it inline — do not import from p3-02)
• Returns (X, y) where X excludes income column, y is the target

main()

1. Parse args
2. Load raw data with load_raw
3. validate_input(X) — raises on schema errors
4. build_pipeline() → fit_and_transform(pipeline, X) → produces processed_arr
5. validate_output(processed_arr) — raises on NaN or shape issues
6. Convert processed_arr to DataFrame using get_feature_names; attach y as last column
7. Write to --output as CSV (index=False)
8. If --save-pipeline given: joblib.dump(pipeline, args.save_pipeline)
9. Print: Processed {n_rows} rows → {n_cols} features. Saved to {output}
10. If --save-pipeline given: print Pipeline saved to {path}

Edge cases:

• If --output directory does not exist, create it with os.makedirs(..., exist_ok=True)

---

src/sagemaker_processor.py

CLI: python src/sagemaker_processor.py [--instance-type ml.m5.large] [--input-s3 s3://...] [--output-s3 s3://...]

upload_input_data(local_path: str, bucket: str, prefix: str) -> str

• Uploads raw.csv to S3; returns S3 URI

submit_processing_job(input_s3: str, output_s3: str, instance_type: str) -> str

• Creates sagemaker.sklearn.SKLearnProcessor(framework_version="1.2-1", ...)
• Calls .run(code="src/main.py", inputs=[ProcessingInput(...)], outputs=[ProcessingOutput(...)], wait=True)
• Returns the processing job name

download_output(job_name: str, local_dir: str) -> None

• Downloads processed CSV from S3 output to local_dir
• Prints: Downloaded processed data to {local_dir}

---

tests/ — what to test

tests/test_pipeline.py

test_pipeline_output_has_no_nan

• Load a 50-row slice of adult.data as DataFrame with correct columns
• Inject np.nan into several cells (simulate ? replacement)
• Build pipeline, fit_and_transform
• Assert not np.isnan(result).any()

test_pipeline_output_shape_is_non_zero

• Same setup as above
• Assert result.shape[0] == 50 (rows preserved)
• Assert result.shape[1] > 14 (OHE expands categorical columns)

test_joblib_roundtrip_preserves_transform

• Fit pipeline on 100-row split
• Save with joblib.dump to tmp file
• Load with joblib.load
• Transform a new 10-row split with the loaded pipeline
• Assert output has no NaN and correct row count

test_local_and_sagemaker_use_same_script

• Assert that src/main.py exists
• Assert the file contains --input (argparse arg used by both local and SageMaker invocations)
• Assert the file contains validate_input call

tests/test_validator.py

test_validate_input_raises_on_missing_column

• Create DataFrame missing the age column
• Assert validate_input(df) raises ValueError containing "age"

test_validate_output_raises_on_nan

• Create a 2D numpy array with one NaN cell
• Assert validate_output(arr) raises ValueError containing "NaN"

---

README.md content

# P3-03 — Feature Engineering

Reusable sklearn preprocessing pipeline for UCI Adult Income. Runs locally and as SageMaker Processing.

## Prerequisites

- Python 3.11+
- AWS credentials — only needed for SageMaker Processing step

## Quick start (local only)

```bash
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
mkdir -p data && curl -o data/raw.csv \
  "https://archive.ics.uci.edu/ml/machine-learning-databases/adult/adult.data"
python src/main.py --input data/raw.csv --output data/processed.csv \
  --save-pipeline pipeline.joblib

What the pipeline does

1. Numeric columns: median imputation → StandardScaler
2. Categorical columns: most_frequent imputation → OneHotEncoder (handle_unknown=ignore)
3. ColumnTransformer combines both branches
4. Validates input schema before fitting, validates output (no NaN) after

Verify no NaN in output

python -c "import pandas as pd; df=pd.read_csv('data/processed.csv'); print('NaN:', df.isna().sum().sum())"

Run as SageMaker Processing Job

cp .env.example .env  # fill in S3_BUCKET and SAGEMAKER_ROLE_ARN
python src/sagemaker_processor.py

Tests

pytest tests/ -v

Teardown

bash scripts/teardown.sh

---

## GUIDE.md content

```markdown
# Guide — P3-03 Feature Engineering

## Why a Pipeline, not manual steps

Without a Pipeline, you must remember to apply the same scaling and encoding at inference time
that you applied at training time. Forget once and your model receives unscaled inputs — it
silently produces wrong predictions. With a fitted Pipeline serialized to joblib, you call
`pipeline.transform(new_data)` at inference and the exact same transformations apply.

## The ColumnTransformer explained

ColumnTransformer applies different transformations to different column subsets and concatenates
the results horizontally. Numeric columns become scaled floats. Categorical columns become
a sparse binary matrix (one column per category value). The total output width is
`len(NUMERIC_FEATURES) + sum(unique_values_per_categorical_column)`.

For UCI Adult, expect roughly 6 numeric + ~100 OHE columns = ~106 features total.

## handle_unknown="ignore" matters

At inference time your model may see a `native-country` value that was not in training data.
With `handle_unknown="ignore"`, OneHotEncoder outputs a zero vector for that value instead of
raising an exception. This is the correct production behavior.

## Validation before and after

**Before:** catches schema drift — if the incoming data has a renamed column, you fail early
with a clear message rather than propagating wrong data into the model.

**After:** catches imputation bugs — if your imputer silently failed to fill NaN (e.g., because
the column dtype was wrong), the post-transform check will catch it.

## SageMaker Processing vs local

SageMaker Processing mounts your input from S3 at `/opt/ml/processing/input/` and collects
your output from `/opt/ml/processing/output/`. Your `main.py` uses `--input` and `--output`
arguments — the SKLearnProcessor passes the mounted paths automatically. The script does not
need to know it is running in a container.

## Interview framing

"I build preprocessing as a serialized sklearn Pipeline so the same transformations apply
identically at training and inference time — no manual step synchronization, no silent
preprocessing drift between training and production."

---

Failure modes to handle

Failure	Where	How to handle
Column with all-NaN values breaks most_frequent imputer	pipeline.py: build_pipeline	SimpleImputer(strategy="most_frequent") returns NaN for all-NaN column — post-transform validate_output will catch it
OHE sees unknown category at inference	pipeline.py	handle_unknown="ignore" on OneHotEncoder outputs zero vector instead of error
Schema mismatch (column renamed upstream)	validator.py: validate_input	Raise ValueError listing the missing columns before any transformation
--output directory does not exist	main.py: main	os.makedirs(dirname, exist_ok=True) before writing CSV
SageMaker Processing Job fails	sagemaker_processor.py	Catch Exception from .run(), print failure reason from job description
joblib.load on wrong Python version	tests/test_pipeline.py note	Document in README: pipeline files are not portable across Python major versions
Post-transform NaN (imputer edge case)	validator.py: validate_output	Raise with message identifying which step likely failed

---

The metric this project measures

What: Output feature count after one-hot encoding (indicator of pipeline correctness)

Format: Integer, printed as Processed {n_rows} rows → {n_cols} features

Target: For UCI Adult Income: approximately 100-110 features (6 numeric + ~100 OHE columns). Exact number depends on unique values in categorical columns in the training split.

Secondary metric: NaN count in output (must be 0), printed as NaN cells in output: 0

---

Cost estimate

Resource	Qty	Rate	Estimated cost
SageMaker Processing Job (ml.m5.large, ~5 min)	1 run	$0.115/hr	~$0.01
S3 storage (raw + processed CSV, ~10 MB)	1 session	$0.023/GB-month	<$0.01
Total per session			~$0.10

Teardown checklist

scripts/teardown.sh must:

• Delete s3://${S3_BUCKET}/p3-03/input/ prefix
• Delete s3://${S3_BUCKET}/p3-03/output/ prefix
• Print Teardown complete.