Issue with Loading Retail Data Model from Wide World Importers into Data Warehouse on Fabric

Hi ahmadammari ,

Unfortunately, it seems that the wwi sample data storage container is no longer available.

At the beginning of "Step 5. Prepare data", please try running the following code to load the wwi sample data provided by Microsoft (Tutorial: Load data using Azure portal & SSMS - Azure Synapse Analytics | Microsoft Learn).

from pyspark.sql.types import *  

fact_sale_schema = StructType([  
    StructField("SaleKey", LongType(), True),  
    StructField("CityKey", IntegerType(), True),  
    StructField("CustomerKey", IntegerType(), True),  
    StructField("BillToCustomerKey", IntegerType(), True),  
    StructField("StockItemKey", IntegerType(), True),  
    StructField("InvoiceDateKey", TimestampType(), True),  
    StructField("DeliveryDateKey", TimestampType(), True),  
    StructField("SalespersonKey", IntegerType(), True),  
    StructField("WWIInvoiceID", IntegerType(), True),  
    StructField("Description", StringType(), True),  
    StructField("Package", StringType(), True),  
    StructField("Quantity", IntegerType(), True),  
    StructField("UnitPrice", DecimalType(18, 2), True),  
    StructField("TaxRate", DecimalType(18, 3), True),  
    StructField("TotalExcludingTax", DecimalType(18, 2), True),  
    StructField("TaxAmount", DecimalType(18, 2), True),  
    StructField("Profit", DecimalType(18, 2), True),  
    StructField("TotalIncludingTax", DecimalType(18, 2), True),  
    StructField("TotalDryItems", IntegerType(), True),  
    StructField("TotalChillerItems", IntegerType(), True),  
    StructField("LineageKey", IntegerType(), True)
])  

dimension_city_schema = StructType([  
    StructField("CityKey", IntegerType(), True),  
    StructField("WWICityID", IntegerType(), True),  
    StructField("City", StringType(), True),  
    StructField("StateProvince", StringType(), True),  
    StructField("Country", StringType(), True),  
    StructField("Continent", StringType(), True),  
    StructField("SalesTerritory", StringType(), True),  
    StructField("Region", StringType(), True),  
    StructField("Subregion", StringType(), True),  
    StructField("Location", StringType(), True),  
    StructField("LatestRecordedPopulation", LongType(), True),  
    StructField("ValidFrom", TimestampType(), True),  
    StructField("ValidTo", TimestampType(), True),  
    StructField("LineageKey", IntegerType(), True)  
])  

dimension_employee_schema = StructType([  
    StructField("EmployeeKey", IntegerType(), True),  
    StructField("WWIEmployeeID", IntegerType(), True),  
    StructField("Employee", StringType(), True),  
    StructField("PreferredName", StringType(), True),  
    StructField("IsSalesperson", IntegerType(), True),  
    StructField("Photo", StringType(), True),  
    StructField("ValidFrom", TimestampType(), True),  
    StructField("ValidTo", TimestampType(), True),  
    StructField("LineageKey", IntegerType(), True)  
])  

dimension_stock_item_schema = StructType([  
    StructField("StockItemKey", IntegerType(), True),  
    StructField("WWIStockItemID", IntegerType(), True),  
    StructField("StockItem", StringType(), True),  
    StructField("Color", StringType(), True),  
    StructField("SellingPackage", StringType(), True),  
    StructField("BuyingPackage", StringType(), True),  
    StructField("Brand", StringType(), True),  
    StructField("Size", StringType(), True),  
    StructField("LeadTimeDays", IntegerType(), True),  
    StructField("QuantityPerOuter", IntegerType(), True),  
    StructField("IsChillerStock", BooleanType(), True),  
    StructField("Barcode", StringType(), True),  
    StructField("TaxRate", DecimalType(18, 3), True),  
    StructField("UnitPrice", DecimalType(18, 2), True),  
    StructField("RecommendedRetailPrice", DecimalType(18, 2), True),  
    StructField("TypicalWeightPerUnit", DecimalType(18, 3), True),  
    StructField("Photo", StringType(), True),  
    StructField("ValidFrom", TimestampType(), True),  
    StructField("ValidTo", TimestampType(), True),  
    StructField("LineageKey", IntegerType(), True)  
])
dimension_customer_schema = StructType([  
    StructField("CustomerKey", LongType(), True),  
    StructField("WWICustomerID", LongType(), True),  
    StructField("Customer", StringType(), True),  
    StructField("BillToCustomer", StringType(), True),  
    StructField("Category", StringType(), True),  
    StructField("BuyingGroup", StringType(), True),  
    StructField("PrimaryContact", StringType(), True),  
    StructField("PostalCode", StringType(), True),  
    StructField("ValidFrom", TimestampType(), True),  
    StructField("ValidTo", TimestampType(), True),  
    StructField("LineageKey", LongType(), True)  
])  

tables = [
    {
        "dist_folder": "fact_sale_1y_full",
        "sorce_folder": "fact_Sale",
        "schema": fact_sale_schema
    },
    {
        "dist_folder": "dimension_city",
        "sorce_folder": "dimension_City",
        "schema": dimension_city_schema
    },
    {
        "dist_folder": "dimension_employee",
        "sorce_folder": "dimension_Employee",
        "schema": dimension_employee_schema
    },
    {
        "dist_folder": "dimension_stock_item",
        "sorce_folder": "dimension_StockItem",
        "schema": dimension_stock_item_schema
    },
    {
        "dist_folder": "dimension_customer",
        "sorce_folder": "dimension_Customer",
        "schema": dimension_customer_schema
    }
]

dist_lakehoues_files = "Files/wwi-raw-data/full/"
source_wasb = "wasbs://[email protected]/v1"
# folders = mssparkutils.fs.ls(source_wasb)

for table in tables:
    schema = table["schema"]
    source_folder=f"{source_wasb}/"+ table["sorce_folder"]
    dist_folder=f"{dist_lakehoues_files}/"+ table["dist_folder"]
    df = spark.read.format("csv").option("delimiter","|").schema(schema).load(source_folder)
    if table["sorce_folder"] == "fact_Sale":
        df = df.where("InvoiceDateKey <= '2000-11-30 00:00:00' AND InvoiceDateKey >= '2000-01-01 00:00:00'")

    df.write.format("parquet").mode("overwrite").save(dist_folder)


## dim date

from pyspark.sql.functions import sequence, explode

start_date = "2000-01-01"  
end_date = "2016-12-31"  
date_df = spark.sql(f"SELECT explode(sequence(to_date('{start_date}'), to_date('{end_date}'), interval 1 day)) as Date")
date_format_df = spark.sql("""
    SELECT 
        `Date`,
        dayofmonth(`Date`) as `DayNumber`,
        cast(dayofmonth(`Date`) as string ) as `Day`,
        date_format(`Date`, "MMMM") as `Month`,
        date_format(`Date`, "MMM") as `ShortMonth`,
        month(`Date`) as `CalendarMonthNumber`,
        concat('CY',year(`Date`),'-',date_format(`Date`, "MMM")) as `CalendarMonthLabel`,
        year(`Date`) as `CalendarYear`,
        concat('CY',year(`Date`)) as `CalendarYearLabel`
    FROM 
        {df}
""",df = date_df)

date_dist_folder = f"{dist_lakehoues_files}/dimension_date"
date_format_df.write.format("parquet").mode("overwrite").save(date_dist_folder)