Project: Building a Retail Data Pipeline

import pandas as pd
import os

# Extract function is already implemented for you 
def extract(store_data, extra_data):
    try:
        extra_df = pd.read_parquet(extra_data)
        merged_df = store_data.merge(extra_df, on = "index")
        print("Merge complete.")
        return merged_df
    except Exception as e:
        print(e)

# Call the extract() function and store it as the "merged_df" variable
merged_df = extract(grocery_sales, "extra_data.parquet")

# Looking at the data
print(merged_df.head)

# Create the transform() function with one parameter: "raw_data"
def transform(raw_data):
    try:
        # Fill missing cells with their column mean values
        # raw_data[['Weekly_Sales', 'CPI', 'Unemployment']].fillna(0, inplace=True)
        raw_data.fillna(
            {
                'Weekly_Sales': raw_data['Weekly_Sales'].mean(),
                'CPI': raw_data['CPI'].mean(),
                'Unemployment': raw_data['Unemployment'].mean(),
            },
            inplace=True
        )
        print("Filled missing values")

        # Convert Date to datetime format
        raw_data['Date'] = pd.to_datetime(raw_data['Date'], format='%Y-%m-%d')
        print("Converted Date column to datetime type")
        # Create Month column
        raw_data["Month"] = raw_data['Date'].dt.month
        print("Month column created.")

        # Drop rows with weekly sales $10,000 or less
        raw_data = raw_data.loc[raw_data['Weekly_Sales'] > 10000, :]
        print("Dropped rows with weekly sales < 10000")
        
        # Filtering unnecessary columns
        raw_data = raw_data[["Store_ID", "Month", "Dept", 
                 "IsHoliday", "Weekly_Sales", "CPI", "Unemployment"]]
        print("Filtered unnecessary columns")

        print("Transform complete.")
        print(raw_data.head)
        return raw_data
        
    except Exception as e:
        print(e)

# Call the transform() function and pass the merged DataFrame
clean_data = transform(merged_df)

# Create the avg_weekly_sales_per_month function that takes in the cleaned data from the last step
def avg_weekly_sales_per_month(clean_data):
    try:
        # Slicing columns
        clean_data = clean_data[["Month", "Weekly_Sales"]]
        print("done column slice")
        
        # Grouping by month, finding the avg weekly sales, reseting the index, rounding to 2 decimals
        clean_data = clean_data.groupby("Month").agg({'Weekly_Sales': 'mean'}).reset_index().round(2)
        print("done group by month and finding avg")

        # Renaming weekly_sales column
        clean_data.rename(columns={'Weekly_Sales': 'Avg_Sales'}, inplace=True)
        
        print(clean_data.head())

        print("Avg monthly sales complete.")
        return clean_data
    except Exception as e:
        print(e)

# Call the avg_weekly_sales_per_month() function and pass the cleaned DataFrame
agg_data = avg_weekly_sales_per_month(clean_data)

# Create the load() function that takes in the cleaned DataFrame and the aggregated one with the paths where they are going to be stored
def load(full_data, full_data_file_path, agg_data, agg_data_file_path):
    # Write your code here
    try:
        # Load full data to csv
        full_data.to_csv(full_data_file_path, index=False)
        print(f'Full data loaded to: {full_data_file_path}')

        #Load agg. data to csv
        agg_data.to_csv(agg_data_file_path, index=False)
        print(f'Agg. data loaded to: {agg_data_file_path}')
        
    except Exception as e:
        print(e)

# Call the load() function and pass the cleaned and aggregated DataFrames with their paths  
load(clean_data, 'clean_data.csv', agg_data, 'agg_data.csv')

# Create the validation() function with one parameter: file_path - to check whether the previous function was correctly executed
def validation(file_path):
    # Write your code here
    try:
        if os.path.exists(file_path):
            print(f'File Path: {file_path} exists in the current directory!')
        else:
            print(f'File Path: {file_path} does not exist!')
    except Exception as e:
        print(e)

# Call the validation() function and pass first, the cleaned DataFrame path, and then the aggregated DataFrame path
validation('clean_data.csv')
validation('agg_data.csv')