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):
    extra_df = pd.read_parquet(extra_data)
    merged_df = store_data.merge(extra_df, on = "index")
    return merged_df

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

# Create the transform() function with one parameter: "raw_data"
def transform(raw_data):
    """
    The Function fills numeric values and creates a "Month" column and removing rows that don't meet the sales threshold
    -----------------------
    Params: 
        raw_data (pandas datafame) : the raw data
    Returns:
        cleaned_data (pandas dataframe) : the cleaned data after applying the transformation
    """
    cleaned_data = pd.DataFrame()
    for column in raw_data.columns:
        if raw_data[column].dtype in [int, float]:
            if raw_data[column].isna().sum() > 0:
                raw_data[column] = raw_data[column].fillna(raw_data[column].mean())
            else:
                raw_data[column] = raw_data[column]
        else:
            raw_data[column] = raw_data[column]
    raw_data['Month'] = raw_data["Date"].dt.month
    cleaned_data = raw_data[raw_data["Weekly_Sales"]> 10000]
    cleaned_data = cleaned_data[["Store_ID","Month","Dept","IsHoliday","Weekly_Sales","CPI","Unemployment"]]
    
    return cleaned_data

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

# 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):
    """
    this function takes a clean data and returns a datafrsme with average weekly sales in each month
    
    ------------------------
    
    Params :
    	clean_data (pandas dataframe) : The clean data
    
    Returns :
    	agg_data (pandas dataframe) : A dataframe with month and average weekly sales
    """
    # Write your code here
    agg_data = clean_data.groupby("Month")["Weekly_Sales"].agg(Avg_Sales = 'mean').reset_index().round(2)
    return agg_data

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

# 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):
    """
    The function saves the data as csv file in the file paths defined
    
    --------------------------
    
    Params:
    	full_data (pandas dataframe) : The clean data.
        full_data_file_path (str) : The clean data file path to be saved in.
        agg_data (pandas dataframe) : The analayized data.
        agg_data_file_path (str) : The analayized data file path.
    """
    full_data.to_csv(full_data_file_path, index = False)
    agg_data.to_csv(agg_data_file_path, index =False)
    print(f"{full_data_file_path} & {agg_data_file_path} are loaded succesfully")

# 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
    if os.path.exists(file_path):
        print(f"{file_path} file saved successfully")
    else:
        raise Exception(f"{file_path} wasn't saved! please reload the file")
    

# 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")