Skip to content
Project: Building a Retail Data Pipeline
  • AI Chat
  • Code
  • Report
  • Walmart is the biggest retail store in the United States. Just like others, they have been expanding their e-commerce part of the business. By the end of 2022, e-commerce represented a roaring $80 billion in sales, which is 13% of total sales of Walmart. One of the main factors that affects their sales is public holidays, like the Super Bowl, Labour Day, Thanksgiving, and Christmas.

    In this project, you have been tasked with creating a data pipeline for the analysis of supply and demand around the holidays, along with conducting a preliminary analysis of the data. You will be working with two data sources: grocery sales and complementary data. You have been provided with the grocery_sales table in PostgreSQL database with the following features:

    grocery_sales

    • "index" - unique ID of the row
    • "Store_ID" - the store number
    • "Date" - the week of sales
    • "Weekly_Sales" - sales for the given store

    Also, you have the extra_data.parquet file that contains complementary data:

    extra_data.parquet

    • "IsHoliday" - Whether the week contains a public holiday - 1 if yes, 0 if no.
    • "Temperature" - Temperature on the day of sale
    • "Fuel_Price" - Cost of fuel in the region
    • "CPI" – Prevailing consumer price index
    • "Unemployment" - The prevailing unemployment rate
    • "MarkDown1", "MarkDown2", "MarkDown3", "MarkDown4" - number of promotional markdowns
    • "Dept" - Department Number in each store
    • "Size" - size of the store
    • "Type" - type of the store (depends on Size column)

    You will need to merge those files and perform some data manipulations. The transformed DataFrame can then be stored as the clean_data variable containing the following columns:

    • "Store_ID"
    • "Month"
    • "Dept"
    • "IsHoliday"
    • "Weekly_Sales"
    • "CPI"
    • ""Unemployment""

    After merging and cleaning the data, you will have to analyze monthly sales of Walmart and store the results of your analysis as the agg_data variable that should look like:

    MonthWeekly_Sales
    1.033174.178494
    2.034333.326579
    ......

    Finally, you should save the clean_data and agg_data as the csv files.

    It is recommended to use pandas for this project.

    Unknown integration
    DataFrameavailable as
    grocery_sales
    variable
    -- Write your SQL query here
    select * from grocery_sales;
    
    This query is taking long to finish...Consider adding a LIMIT clause or switching to Query mode to preview the result.
    import pandas as pd
    import os
    
    # Start here...
    file_paths = {
        'clean_data': 'clean_data.csv',
        'agg_data': 'agg_data.csv'
    }
    
    def extract(grocery_sales: pd.DataFrame, parquet_file_name: str) -> pd.DataFrame:
        extra_data_df = pd.read_parquet(f'{parquet_file_name}.parquet')
        merged_df = grocery_sales.merge(extra_data_df)
        return merged_df
        
    def transform(merged_df: pd.DataFrame) -> pd.DataFrame:
        # Fill missing numerical values with 0 (simple method)
        merged_df.fillna(0, inplace=True)
        
        # Convert 'Date' column to datetime format
        merged_df['Date'] = pd.to_datetime(merged_df['Date'], errors='coerce')  # Added errors='coerce'
        
        # Extract month from the Date column and create a new column "Month"
        merged_df['Month'] = merged_df['Date'].dt.month
        
        # Keep rows where the weekly sales are over $10,000
        filtered_df = merged_df[merged_df['Weekly_Sales'] > 10000]
        
        # Drop unnecessary columns
        # Assuming unnecessary columns are those not listed in the provided variables
        necessary_columns = [
            'Store_ID', 'Month', 'Dept', 'IsHoliday', 'Weekly_Sales', 'CPI', 'Unemployment']
        
        clean_data = filtered_df[necessary_columns]
        
        return clean_data
    
    def avg_monthly_sales(clean_data_df: pd.DataFrame) -> pd.DataFrame:
        # Group by month and calculate average sales
        agg_data = clean_data_df.groupby('Month')['Weekly_Sales'].mean().reset_index()
    
        # Round Avg_Sales to 2 decimals
        agg_data['Avg_Sales'] = agg_data['Weekly_Sales'].round(2)
        
        # Select only 'Month' and 'Avg_Sales'
    
        return agg_data[['Month', 'Avg_Sales']]
    
    def load(
            clean_data_df: pd.DataFrame, 
            clean_data_path: str, 
            agg_data_df: pd.DataFrame, 
            agg_data_path: str
        ) -> None:
        clean_data_df.to_csv(clean_data_path, index=False)
        agg_data_df.to_csv(agg_data_path, index=False)
        
    def validation(file_path: str) -> None:
        if not os.path.exists(file_path):  # Corrected 'exits' to 'exists'
            print(f"Error: CSV file '{file_path}' not found.")
        else:
            print(f'CSV file {file_path} exists')
    
    
    merged_df = extract(grocery_sales=grocery_sales, parquet_file_name='extra_data')
    clean_data = transform(merged_df)
    agg_data = avg_monthly_sales(clean_data)
    load(clean_data_df=clean_data, 
         clean_data_path=file_paths['clean_data'], 
         agg_data_df=agg_data,
         agg_data_path=file_paths['agg_data']
        )
    validation(file_path=file_paths['clean_data'])