From 31b68c612713dada1b667bf2ab2dcda639be238e Mon Sep 17 00:00:00 2001 From: baraah Date: Thu, 28 May 2026 21:12:41 +0200 Subject: [PATCH] Finish week4 assignment with Azure integration --- AI_ASSIST.md | 19 ++++++++++- main.py | 2 +- src/clean.py | 46 +++++++++++++++++++++++++++ src/ingest.py | 54 +++++++++++++++++++++++++++++-- src/report.py | 82 ++++++++++++++++++++++++++++++++++++++++++++++-- src/transform.py | 10 ++++++ 6 files changed, 207 insertions(+), 6 deletions(-) diff --git a/AI_ASSIST.md b/AI_ASSIST.md index a50dbac..78bff6e 100644 --- a/AI_ASSIST.md +++ b/AI_ASSIST.md @@ -3,9 +3,20 @@ ## The prompt I gave - +I needed help implementing the final part of Task 7 in my Pandas + Azure data pipeline assignment. ## The code it suggested +local_file = output_dir / "customer_summary.parquet" + +blob = container.get_blob_client("customer_summary.parquet") +downloaded = blob.download_blob().readall() + +remote_df = pd.read_parquet(io.BytesIO(downloaded)) +local_df = pd.read_parquet(local_file) +assert len(remote_df) == len(local_df), "Row count mismatch!" + +logging.info("Round-trip verification passed ✔") +logging.info(f"Uploaded {len(parquet_files)} files to {container_name}") ```python # Paste the relevant code the LLM suggested here. ``` @@ -13,7 +24,13 @@ ## What I changed and why +I did not change the main logic because the round-trip verification approach was already correct. +I kept the assert statement because it is a simple and effective way to validate data consistency. +I kept logging as it helps track successful execution in the pipeline. ## Did it work? +The code worked successfully and uploaded files to Azure and downloaded customer_summary.parquet back from Blob Storage , also verified that the row counts matched without any errors + +I learned that the key part of this step is not just uploading files, but ensuring data integrity between local and cloud storage. diff --git a/main.py b/main.py index 9f17efc..4a8c402 100644 --- a/main.py +++ b/main.py @@ -13,7 +13,7 @@ OUTPUT_DIR = Path("output") # TODO (Task 7): replace with your GitHub username before running the pipeline. -GITHUB_USERNAME = "" +GITHUB_USERNAME = "thebaraah" def run() -> None: diff --git a/src/clean.py b/src/clean.py index b4036fd..67e826b 100644 --- a/src/clean.py +++ b/src/clean.py @@ -10,19 +10,65 @@ def load_and_explore(data_dir: Path) -> tuple[pd.DataFrame, pd.DataFrame]: # TODO: Read messy_sales.csv and messy_customers.csv with pd.read_csv(). # TODO: For each DataFrame call .info(), .describe(), .head(20), and .isna().sum(). # TODO: Log what you discover (e.g. which columns have nulls, any suspicious values). + sales_path = data_dir / "messy_sales.csv" + customers_path = data_dir / "messy_customers.csv" + + sales = pd.read_csv(sales_path) + customers = pd.read_csv(customers_path) + + logging.info("=== SALES INFO ===") + logging.info(sales.info()) + + logging.info("=== CUSTOMERS INFO ===") + logging.info(customers.info()) + + logging.info("=== SALES MISSING VALUES ===") + logging.info(sales.isna().sum()) + + logging.info("=== CUSTOMERS MISSING VALUES ===") + logging.info(customers.isna().sum()) + + logging.info("=== SALES SUMMARY ===") + logging.info(sales.describe()) + + logging.info("=== CUSTOMERS SAMPLE ===") + logging.info(sales.head(20)) + + return sales, customers raise NotImplementedError("Task 2: implement load_and_explore") def clean_sales(sales: pd.DataFrame) -> pd.DataFrame: """Task 3: Clean the sales DataFrame using vectorized Pandas operations.""" + df = sales.copy() # TODO: Normalize product_name with .str.strip().str.title(). + df["product_name"] = df["product_name"].str.strip().str.title() + # TODO: Normalize customer_email with .str.lower().str.strip(). + df["customer_email"] = df["customer_email"].str.lower().str.strip() # TODO: Convert price to numeric with pd.to_numeric(errors="coerce"). + df["price"] = pd.to_numeric(df["price"], errors="coerce") # TODO: Parse date with pd.to_datetime(errors="coerce"). + df["date"] = pd.to_datetime(df["date"], errors="coerce") # TODO: Drop rows where product_name is missing. + df = df[df["product_name"].notna()] + df = df[df["price"].notna()] + df = df[df["price"] >= 0] + df = df[df["quantity"] > 0] + df = df[df["date"].notna()] + # TODO: Drop rows where price is negative. + df = df.drop_duplicates(subset="transaction_id", keep="first") # TODO: Drop rows where quantity is zero. + # TODO: Drop rows where date is NaT (invalid after parsing). # TODO: Remove duplicate transactions: .drop_duplicates(subset="transaction_id", keep="first"). # TODO: Decide what to do with outlier prices (clip, flag, or leave) and add a comment explaining why. + q99 = df["price"].quantile(0.99) + df["price"] = df["price"].clip(upper=q99) + + logging.info(f"Cleaned sales rows: {len(df)}") + + return df + raise NotImplementedError("Task 3: implement clean_sales") diff --git a/src/ingest.py b/src/ingest.py index 01fe28f..4eabe06 100644 --- a/src/ingest.py +++ b/src/ingest.py @@ -15,20 +15,70 @@ def download_inputs(data_dir: Path) -> None: """Task 1: Download input CSV files from Azure Blob Storage.""" # TODO: Create a BlobServiceClient using DefaultAzureCredential and ACCOUNT_URL. + #Create Azure client + credential = DefaultAzureCredential() + service = BlobServiceClient( + account_url=ACCOUNT_URL, + credential=credential + ) # TODO: Get a container client for SOURCE_CONTAINER. + container_client = service.get_container_client(SOURCE_CONTAINER) # TODO: For each filename in FILES, download the blob and write it to data_dir/. # TODO: Log a message for each downloaded file. - raise NotImplementedError("Task 1: implement download_inputs") + data_dir.mkdir(parents=True, exist_ok=True) + for filename in FILES: + blob = container_client.get_blob_client(filename) + + file_path = data_dir / filename + + with open(file_path, "wb") as f: + f.write(blob.download_blob().readall()) + + logging.info(f"Downloaded {filename} -> {file_path}") + def upload_outputs(output_dir: Path, github_username: str) -> None: """Task 7 (extra credit): Upload Parquet outputs to Azure and verify the round-trip.""" + logging.info("Starting upload to Azure...") container_name = f"week4-{github_username}" # EXTRA CREDIT — implement this after Tasks 2–6 are working. # TODO: Create a BlobServiceClient using DefaultAzureCredential and ACCOUNT_URL. + + credential = DefaultAzureCredential() + service = BlobServiceClient( + account_url=ACCOUNT_URL, + credential=credential + ) # TODO: Get (or create) the container named container_name. + container_client = service.get_container_client(container_name) + try: + container_client.create_container() + except Exception as e: + logging.warning(f"Container might already exist: {e}") + # TODO: Upload every .parquet file in output_dir to the container. + parquet_files = list(output_dir.glob("*.parquet")) + + for file_path in parquet_files: + blob = container_client.get_blob_client(file_path.name) + + with open(file_path, "rb") as data: + blob.upload_blob(data, overwrite=True) + + logging.info(f"Uploaded {file_path.name}") # TODO: Download customer_summary.parquet back and assert its row count matches the local file. # TODO: Log the container name and number of files uploaded. - raise NotImplementedError("Task 7: implement upload_outputs") + local_file = output_dir / "customer_summary.parquet" + + blob = container_client.get_blob_client("customer_summary.parquet") + downloaded = blob.download_blob().readall() + + remote_df = pd.read_parquet(io.BytesIO(downloaded)) + local_df = pd.read_parquet(local_file) + + assert len(remote_df) == len(local_df), "Row count mismatch!" + + logging.info("Round-trip verification passed ✔") + logging.info(f"Uploaded {len(parquet_files)} files to {container_name}") diff --git a/src/report.py b/src/report.py index a002b7b..fe245a1 100644 --- a/src/report.py +++ b/src/report.py @@ -3,18 +3,71 @@ from pathlib import Path import pandas as pd +import matplotlib +matplotlib.use("Agg") + +import matplotlib.pyplot as plt def build_reports(enriched: pd.DataFrame) -> dict[str, pd.DataFrame]: """Task 5: Build four summary tables using groupby and named aggregations.""" + df = enriched.copy() + # TODO: Add a week column using .dt.isocalendar().week. + df["week"] = df["date"].dt.isocalendar().week + # TODO: Build weekly_revenue: group by week and region, columns week/region/total_revenue/order_count. + weekly_revenue = ( + df.groupby(["week", "region"]) + .agg( + total_revenue=("price", lambda x: (x * df.loc[x.index, "quantity"]).sum()), + order_count=("transaction_id", "count") + ) + .reset_index() + ) + # TODO: Build customer_summary: group by customer_email, columns customer_email/customer_name/ # region/loyalty_tier/total_spent/avg_order/order_count. # Use ("customer_name", "first") to pick the constant-per-group string columns. + customer_summary = ( + df.groupby("customer_email") + .agg( + customer_name=("customer_name", "first"), + region=("region", "first"), + loyalty_tier=("loyalty_tier", "first"), + total_spent=("price", lambda x: (x * df.loc[x.index, "quantity"]).sum()), + avg_order=("price", lambda x: (x * df.loc[x.index, "quantity"]).mean()), + order_count=("transaction_id", "count") + ) + .reset_index() + ) # TODO: Build category_performance: group by category, columns category/total_revenue/order_count. + category_performance = ( + df.groupby("category") + .agg( + total_revenue=("price", lambda x: (x * df.loc[x.index, "quantity"]).sum()), + order_count=("transaction_id", "count") + ) + .reset_index() + ) # TODO: Build loyalty_analysis: group by loyalty_tier, columns loyalty_tier/avg_spent/customer_count. - raise NotImplementedError("Task 5: implement build_reports") + loyalty_analysis = ( + df.groupby("loyalty_tier") + .agg( + avg_spent=("price", lambda x: (x * df.loc[x.index, "quantity"]).mean()), + customer_count=("customer_email", "nunique") + ) + .reset_index() + ) + + logging.info("Reports built successfully") + + return { + "weekly_revenue": weekly_revenue, + "customer_summary": customer_summary, + "category_performance": category_performance, + "loyalty_analysis": loyalty_analysis, + } def write_outputs(reports: dict[str, pd.DataFrame], output_dir: Path) -> None: @@ -22,10 +75,35 @@ def write_outputs(reports: dict[str, pd.DataFrame], output_dir: Path) -> None: output_dir.mkdir(exist_ok=True) # TODO: Write reports["weekly_revenue"] to weekly_revenue.csv with index=False. + reports["weekly_revenue"].to_csv(output_dir / "weekly_revenue.csv", index=False) + + reports["customer_summary"].to_parquet( + output_dir / "customer_summary.parquet", + index=False + ) + + reports["category_performance"].to_csv( + output_dir / "category_performance.csv", + index=False + ) # TODO: Write reports["customer_summary"] to customer_summary.parquet with index=False. + cat = reports["category_performance"].sort_values( + "total_revenue", + ascending=False + ) + # TODO: Write reports["category_performance"] to category_performance.csv with index=False. + plt.figure() + cat.plot( + kind="bar", + x="category", + y="total_revenue", + title="Revenue by category" + ) # TODO: Sort category_performance by total_revenue descending. + plt.savefig(output_dir / "category_revenue.png", bbox_inches="tight") + + logging.info(f"Outputs written to {output_dir}") # TODO: Plot a bar chart (x="category", y="total_revenue") and save to category_revenue.png # using plt.savefig(output_dir / "category_revenue.png", bbox_inches="tight"). # Use matplotlib.use("Agg") before importing pyplot for headless environments. - raise NotImplementedError("Task 6: implement write_outputs") diff --git a/src/transform.py b/src/transform.py index 18f82e5..5468bfc 100644 --- a/src/transform.py +++ b/src/transform.py @@ -6,8 +6,18 @@ def join_customers(sales: pd.DataFrame, customers: pd.DataFrame) -> pd.DataFrame: """Task 4: Normalize join keys, merge, and add a derived boolean flag.""" + sales = sales.copy() + customers = customers.copy() # TODO: Normalize customer_email in both DataFrames with .str.lower().str.strip(). + sales["customer_email"] = sales["customer_email"].str.lower().str.strip() + customers["customer_email"] = customers["customer_email"].str.lower().str.strip() # TODO: Merge sales with customers on customer_email using an inner join. + df = sales.merge(customers, on="customer_email", how="inner") # TODO: Add a vectorized boolean column is_high_value: True where price * quantity >= 150. # TODO: (Optional hands-on) Try a left join instead and inspect rows where customer_name is NaN. + df["is_high_value"] = (df["price"] * df["quantity"]) >= 150 + + logging.info(f"Merged rows: {len(df)}") + + return df raise NotImplementedError("Task 4: implement join_customers")