Practice Project 3

End-to-End Data Wrangling Pipeline

Extract data from multiple formats (CSV + JSON), clean nulls and anomalies, map joins on relational keys, perform wide-to-long table reshaping, and export reusable pipeline modules.

Domain / Environment

Customer Analytics / Conda VM

Difficulty

Intermediate (3/5)

Course Module

Data Collection & Wrangling

Deliverables

Cleaned dataset file & Verification Script

1. System Architecture & Process Workflow

The diagram below displays the ETL (Extract, Transform, Load) pipelines architecture. Order logs (CSV) and customer profiles (JSON) are parsed into temporary dataframes. A pipeline function joins tables on matching IDs, handles missing values and outliers, shapes tables via pivots, and saves the cleaned dataset.

2. Part 1: Step-by-Step Action Items & Key Execution Steps

STEP 1

Activate Python Virtual Environment

Point the terminal execution environment to the course conda sandbox environment.

$ conda activate ds_ai_ml

This action redirects active library pathways to the virtual environment directory folders.

STEP 2

Create Project Folders inside Linux VM

Create a dedicated folder for the project files inside your guest VM home folder directory.

$ mkdir -p ~/Projects/data_wrangling && cd ~/Projects/data_wrangling

This creates the project workspace and shifts the active terminal context into it.

STEP 3

Create Raw Orders CSV Log

Create a CSV dataset containing raw transaction logs, item counts, and pricing.

$ nano orders.csv

This opens nano editor. Paste the CSV data template from Part 2, press **Ctrl + O** and **Enter** to save, and **Ctrl + X** to exit.

STEP 4

Create Raw Customer Profiles JSON File

Create a JSON file containing user profile records, email listings, and signup dates.

$ nano customers.json

This opens nano editor. Paste the JSON template from Part 2, press **Ctrl + O** and **Enter** to save, and **Ctrl + X** to exit.

STEP 5

Create pipeline script file in VS Code

Launch VS Code and create a new script file inside the project workspace folder.

Launch VS Code via terminal "code ." -> Right-click in explorer tree -> click New File -> Type: wrangle_pipeline.py -> Press Enter

This registers an empty file `wrangle_pipeline.py` inside the active directory editor workspace.

STEP 6

Load Pipeline compilation logic

Paste the data cleaning and transformation code into the newly created python script file.

Click wrangle_pipeline.py -> Paste python code from Part 2 below -> Save file via Ctrl + S

This writes the ETL transformation commands to the code file.

STEP 7

Execute pipeline logic script

Run the validation script using the python engine to compile the datasets.

$ python wrangle_pipeline.py

This runs the pipeline, merging the sources, cleaning anomalies, pivoting data, and saving the output to `cleaned_customer_data.csv`.

3. Operational Pipeline Architecture

The flowchart below outlines the data wrangling pipeline. It details the steps from extracting orders and customer profiles to data cleansing, merging tables on relational keys, reshaping, and exporting the final dataset.

4. Part 2: Complete Deliverable Assets & Production Templates

To run the wrangling pipeline, we need orders CSV data, customer JSON data, and python script logic. Below is a line-by-line explanation of the code, followed by the combined template files.

Step-by-Step Code Construction

Lines 1 - 4

Import tabular packages

Include system OS and Pandas libraries in the code script.

import pandas as pd import numpy as np import os

These imports check system paths and load Pandas/NumPy packages for data manipulation.

Lines 5 - 12

Define loading paths & load datasets

Locate the files and read the raw datasets into Pandas DataFrames.

df_orders = pd.read_csv("orders.csv") df_customers = pd.read_json("customers.json") print("Orders row shape:", df_orders.shape) print("Customers row shape:", df_customers.shape)

This parses paths and reads CSV and JSON data into memory, tracking their sizes.

Lines 13 - 22

Clean anomalies and duplicates

Impute null values and remove duplicate rows from the orders database.

df_orders = df_orders.drop_duplicates(subset=["Order_ID"]) df_orders["Spend"] = df_orders["Spend"].fillna(0.0) df_orders = df_orders[df_orders["Spend"] >= 0]

This dedupes order records, replaces null spending values with zero, and filters out negative transactions.

Lines 23 - 35

Map relational joins and engineer features

Join the customer profiles and orders tables, and create derived columns.

df_merged = pd.merge(df_orders, df_customers, on="Customer_ID", how="inner") df_merged["Spend_Per_Age"] = df_merged["Spend"] / df_merged["Age"] df_merged["Tenure_Years"] = 2026 - pd.to_datetime(df_merged["Signup_Date"]).dt.year

This executes an inner join on customer IDs and creates two derived columns: spend-per-age and membership tenure.

Lines 36 - 45

Pivot tables and export final datasets

Pivot the tables by customer category and save the output as a CSV file.

pivot_df = df_merged.pivot_table(index="Customer_ID", columns="Category", values="Spend", aggfunc="sum").fillna(0.0) pivot_df.to_csv("cleaned_customer_data.csv") print("Data pipeline executed successfully. Output saved to cleaned_customer_data.csv")

This aggregates spend by product category for each customer, shapes the table, and exports the final dataset.

Production templates

1. Sample Dataset 1 (Save as ~/Projects/data_wrangling/orders.csv):

Order_ID,Customer_ID,Spend,Category 1001,101,150.00,Electronics 1002,102,45.50,Furniture 1003,101,89.99,Electronics 1004,103,12.50,Office Supplies 1005,104,,Furniture 1006,102,-50.00,Electronics 1007,101,150.00,Electronics 1008,105,210.00,Furniture 1009,103,4.99,Office Supplies 1010,104,120.00,Office Supplies

2. Sample Dataset 2 (Save as ~/Projects/data_wrangling/customers.json):

[ {"Customer_ID": 101, "Age": 28, "Signup_Date": "2024-01-15"}, {"Customer_ID": 102, "Age": 34, "Signup_Date": "2023-06-20"}, {"Customer_ID": 103, "Age": 45, "Signup_Date": "2025-02-10"}, {"Customer_ID": 104, "Age": 22, "Signup_Date": "2024-11-05"}, {"Customer_ID": 105, "Age": 50, "Signup_Date": "2022-03-30"} ]

3. Python script (Save as ~/Projects/data_wrangling/wrangle_pipeline.py):

# wrangle_pipeline.py - Reusable data wrangling pipeline function import pandas as pd import numpy as np import os def wrangle_pipeline(orders_path, customers_path): print("=== Launching Data Wrangling Pipeline ===") # 1. Extraction checks if not os.path.exists(orders_path) or not os.path.exists(customers_path): print("ERROR: Source data files are missing!") return None df_orders = pd.read_csv(orders_path) df_customers = pd.read_json(customers_path) print(f"Extracted: {len(df_orders)} order logs, {len(df_customers)} customer profiles.") # 2. Cleaning Orders print("\n[Cleaning Orders Data]") # Deduplicate orders initial_len = len(df_orders) df_orders = df_orders.drop_duplicates(subset=["Order_ID"]) print(f" - Removed {initial_len - len(df_orders)} duplicate orders.") # Missing values handling null_count = df_orders["Spend"].isnull().sum() df_orders["Spend"] = df_orders["Spend"].fillna(0.0) print(f" - Imputed {null_count} null spend fields with 0.0.") # Outlier handling: Filter negative spends neg_spends = df_orders[df_orders["Spend"] < 0] df_orders = df_orders[df_orders["Spend"] >= 0] print(f" - Filtered out {len(neg_spends)} negative spend records.") # 3. Relational Merge print("\n[Merging Datasets]") df_merged = pd.merge(df_orders, df_customers, on="Customer_ID", how="inner") print(f" - Merged dataset dimensions: {df_merged.shape}") # 4. Feature Engineering print("\n[Engineering Derived Features]") df_merged["Spend_Per_Age"] = df_merged["Spend"] / df_merged["Age"] current_year = 2026 df_merged["Signup_Year"] = pd.to_datetime(df_merged["Signup_Date"]).dt.year df_merged["Tenure_Years"] = current_year - df_merged["Signup_Year"] print(" - Features compiled: Spend_Per_Age and Tenure_Years") # 5. Reshaping (Pivot table) print("\n[Reshaping Dimensions via Pivot]") pivot_df = df_merged.pivot_table( index=["Customer_ID", "Age", "Tenure_Years"], columns="Category", values="Spend", aggfunc="sum" ).fillna(0.0).reset_index() print(" - Table pivoted from wide-to-long transaction structure.") # Automated schema sanity check validations print("\n[Running automated data checks]") assert (pivot_df["Age"] < 0).sum() == 0, "Validation Error: Negative ages found." print(" - Quality Checks completed. Data integrity validated.") return pivot_df if __name__ == "__main__": orders_csv = "orders.csv" customers_json = "customers.json" output_df = wrangle_pipeline(orders_csv, customers_json) if output_df is not None: export_file = "cleaned_customer_data.csv" output_df.to_csv(export_file, index=False) print(f"\nPipeline successfully logged database records to: {export_file}") print("Head Output:") print(output_df.head()) print("=== Data Wrangling Project Successfully Complete! ===")

5. Deliverables Summary

Verify that the following configurations and outputs exist inside your project workspace.

Created Files / Templates

~/Projects/data_wrangling/orders.csv - Raw order transactional log CSV.
~/Projects/data_wrangling/customers.json - Customer details JSON.
~/Projects/data_wrangling/wrangle_pipeline.py - Reusable pipeline script file.
~/Projects/data_wrangling/cleaned_customer_data.csv - Final export dataset.

Verification Artifacts / Execution Proof

Logs showing duplicate orders dropped successfully (ID 1007 dropped).
Negative spends filtered out (order ID 1006 with -50 spend dropped).
Null value imputation checked (empty spend field replaced with 0.0).
Validation checks running without throwing schema assert errors.

6. Closing Explanation: Why We Did This & What It Accomplishes

Architectural Intent & Operational Impact

Why We Did This

Raw data from multiple sources is often messy, so combining extraction and cleaning into one pipeline makes it repeatable.
Feature engineering maps row-level metrics to business values, making downstream modeling easier.
Automated sanity checks flag data anomalies early, before they affect downstream analytics.

What This Accomplishes

Integrates CSV and JSON records into a single dataset.
Imputes missing records and flags out-of-bounds metrics.
Builds a reusable pipeline function that can process new data blocks consistently.