--- title: "Feature Engineering" description: "A single sklearn pipeline that works unchanged as SageMaker Processing is a direct, practical answer to 'how do you avoid train/serve skew' — a question that..." source: "https://confidentprep.com/paths/ml-engineering-on-aws/feature-engineering/" path: "ML Engineering on AWS" repository: "https://github.com/confidentprep14-web/ai-engineering-projects/tree/main/feature-engineering" token_estimate: 3912 --- # Feature Engineering ## Overview A reusable sklearn preprocessing pipeline for the same UCI Adult Income data, built to run locally and as a SageMaker Processing job without modification. This is the project that makes the next ones honest — if features are computed differently in training than at serving time, every model after this is quietly wrong in production. ## What to do **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 ```bash # 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 ```dotenv # 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):** ```python 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 ```markdown # 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 ```bash python -c "import pandas as pd; df=pd.read_csv('data/processed.csv'); print('NaN:', df.isna().sum().sum())" ``` ## Run as SageMaker Processing Job ```bash cp .env.example .env # fill in S3_BUCKET and SAGEMAKER_ROLE_ARN python src/sagemaker_processor.py ``` ## Tests ```bash pytest tests/ -v ``` ## Teardown ```bash 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.` ## Source code A reference implementation of Feature Engineering — README, guide, source and tests — is at https://github.com/confidentprep14-web/ai-engineering-projects/tree/main/feature-engineering.