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://wideworldimporters@sqldwholdata.blob.core.windows.net/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)
