End-to-end retail ETL pipeline using Medallion Architecture (Bronze/Silver/Gold) with TSQL, PySpark, and Airflow for inventory, sales, and supplier data processing
Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.
Quelldateien prüfen
Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
End-to-end retail ETL pipeline using Medallion Architecture (Bronze/Silver/Gold) with TSQL, PySpark, and Airflow for inventory, sales, and supplier data processing
triggers
["build a retail data warehouse with medallion architecture","create bronze silver gold layers for retail analytics","set up retail ETL pipeline with inventory tracking","implement medallion architecture for sales data","process retail data with bronze silver gold pattern","design data warehouse for hypermarket or retail business","transform retail sales and inventory data by layers","orchestrate retail ETL with airflow and spark"]
This project implements a production-grade Medallion Architecture ETL pipeline for retail/hypermarket data, handling complex business logic like inventory shrinkage, meat/poultry recipe conversions, supplier rebate tiers, and multi-branch sales consolidation. The architecture follows three data quality layers:
Bronze Layer: Raw data ingestion from CSV sources (sales, stock, products)
Silver Layer: Cleaned, standardized, and business-rule-applied data
for script in sql_scripts/01_bronze_*.sql sql_scripts/02_bronze_*.sql sql_scripts/03_bronze_*.sql sql_scripts/04_bronze_*.sql; do
sqlcmd -S localhost -U sa -P $SQL_SA_PASSWORD -i "$script"done
Key Architecture Patterns
Bronze Layer (Raw Ingestion)
Purpose: Land raw data with minimal transformation. Add audit columns only.
-- Calculate dynamic rebate percentages based on purchase volumeCREATEPROCEDURE gold.usp_CalculateSupplierRebates
ASBEGININSERT INTO gold.SupplierRebates (
SupplierID,
TotalPurchaseAmount,
RebateTier,
RebatePercent,
RebateAmount
)
SELECT
SupplierID,
SUM(TotalAmount) AS TotalPurchaseAmount,
CASEWHENSUM(TotalAmount) >=100000THEN'Platinum'WHENSUM(TotalAmount) >=50000THEN'Gold'WHENSUM(TotalAmount) >=25000THEN'Silver'ELSE'Bronze'ENDAS RebateTier,
CASEWHENSUM(TotalAmount) >=100000THEN5.0WHENSUM(TotalAmount) >=50000THEN3.0WHENSUM(TotalAmount) >=25000THEN1.5ELSE0.0ENDAS RebatePercent,
SUM(TotalAmount) *CASEWHENSUM(TotalAmount) >=100000THEN0.05WHENSUM(TotalAmount) >=50000THEN0.03WHENSUM(TotalAmount) >=25000THEN0.015ELSE0.0ENDAS RebateAmount
FROM silver.Sales s
INNERJOIN silver.Products p ON s.ProductID = p.ProductID
GROUPBY SupplierID;
END;
Inventory Shrinkage Detection
-- Identify products with abnormal shrinkageSELECT
p.ProductName,
p.Category,
st.BranchID,
st.ExpectedStock,
st.StockQuantity AS ActualStock,
((st.ExpectedStock - st.StockQuantity) *100.0) / st.ExpectedStock AS ShrinkagePercent
FROM silver.Stock st
INNERJOIN silver.Products p ON st.ProductID = p.ProductID
WHERE st.ExpectedStock >0AND ((st.ExpectedStock - st.StockQuantity) *100.0) / st.ExpectedStock >5.0-- >5% shrinkage thresholdORDERBY ShrinkagePercent DESC;
Data Quality Checks
Validation Queries
-- Check for duplicate sales recordsSELECT SaleID, COUNT(*) AS Duplicates
FROM bronze.Sales
GROUPBY SaleID
HAVINGCOUNT(*) >1;
-- Validate price consistencySELECT
p.ProductID,
p.ProductName,
COUNT(DISTINCT s.UnitPrice) AS PriceVariations
FROM silver.Products p
INNERJOIN silver.Sales s ON p.ProductID = s.ProductID
GROUPBY p.ProductID, p.ProductName
HAVINGCOUNT(DISTINCT s.UnitPrice) >3; -- More than 3 price points-- Check for negative stockSELECT ProductID, BranchID, StockQuantity
FROM silver.Stock
WHERE StockQuantity <0;
-- Data completeness metricsSELECT'Products'AS TableName,
COUNT(*) AS TotalRows,
SUM(CASEWHEN ProductName ISNULLTHEN1ELSE0END) AS NullProductNames,
SUM(CASEWHEN UnitPrice ISNULLTHEN1ELSE0END) AS NullPrices
FROM silver.Products;
Troubleshooting
Common Issues
Issue: BULK INSERT fails with permission error
-- Solution: Grant read permissions to SQL Server service account-- Or use OPENROWSET with explicit credentialsINSERT INTO bronze.Products
SELECT*FROM OPENROWSET(
BULK '/data/000.Hypermarket Products.csv',
FORMATFILE ='/data/products_format.xml',
ERRORFILE ='/logs/errors.txt'
) AS DataFile;
-- Check for missing RecipeYield in Products tableSELECT ProductID, ProductName, Category, RecipeYield
FROM bronze.Products
WHERE Category IN ('Meat & Poultry', 'Seafood')
AND RecipeYield ISNULL;
-- Fix: Set default yield to 1.0UPDATE bronze.Products
SET RecipeYield =1.0WHERE RecipeYield ISNULL;
Issue: Silver layer procedure times out on large datasets
-- Solution: Add batch processing with cursor or temp tablesCREATEPROCEDURE silver.usp_TransformSalesBatch
@BatchSizeINT=10000ASBEGINDECLARE@MinIDINT, @MaxIDINT;
SELECT@MinID=MIN(SaleID), @MaxID=MAX(SaleID) FROM bronze.Sales;
WHILE @MinID<=@MaxIDBEGININSERT INTO silver.Sales (...)
SELECT ...
FROM bronze.Sales
WHERE SaleID BETWEEN@MinIDAND (@MinID+@BatchSize-1);
SET@MinID=@MinID+@BatchSize;
END;
END;
Issue: Gold aggregates not updating incrementally
-- Solution: Implement incremental load with watermarkCREATE TABLE gold.ETL_Watermark (
TableName NVARCHAR(100),
LastProcessedDate DATETIME2
);
CREATEPROCEDURE gold.usp_IncrementalInventoryMetrics
ASBEGINDECLARE@LastRun DATETIME2;
SELECT@LastRun= LastProcessedDate FROM gold.ETL_Watermark WHERE TableName ='InventoryMetrics';
-- Delete and recalculate only changed dataDELETEFROM gold.InventoryTurnover
WHEREMonth>= DATEPART(MONTH, @LastRun);
INSERT INTO gold.InventoryTurnover (...)
SELECT ...
FROM silver.Sales
WHERE SaleDate >=@LastRun;
-- Update watermarkUPDATE gold.ETL_Watermark
SET LastProcessedDate = GETDATE()
WHERE TableName ='InventoryMetrics';
END;
Performance Optimization
-- Add indexes for Bronze layer queriesCREATE CLUSTERED INDEX IX_Sales_SaleID ON bronze.Sales(SaleID);
CREATE NONCLUSTERED INDEX IX_Sales_ProductID ON bronze.Sales(ProductID);
CREATE NONCLUSTERED INDEX IX_Sales_SaleDate ON bronze.Sales(SaleDate);
-- Partition Gold tables by month for faster queriesCREATEPARTITIONFUNCTION pf_MonthPartition (INT)
ASRANGERIGHTFORVALUES (1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12);
CREATEPARTITION SCHEME ps_MonthPartition
ASPARTITION pf_MonthPartition ALLTO ([PRIMARY]);
CREATE TABLE gold.InventoryTurnover (
...
MonthINT
) ON ps_MonthPartition(Month);
-- Enable query store for performance monitoringALTER DATABASE RetailDataWarehouse SET QUERY_STORE =ON;
Integration with BI Tools
Power BI Connection
-- Create view optimized for Power BICREATEVIEW gold.vw_SalesDashboard ASSELECT
s.SaleDate,
p.ProductName,
p.Category,
b.BranchName,
s.Quantity,
s.UnitPrice,
s.TotalAmount,
i.TurnoverRatio,
i.ShrinkagePercent
FROM gold.InventoryTurnover i
INNERJOIN silver.Sales s ON i.ProductID = s.ProductID AND i.BranchID = s.BranchID
INNERJOIN silver.Products p ON s.ProductID = p.ProductID
INNERJOIN silver.Branches b ON s.BranchID = b.BranchID;
-- Grant read-only access to BI service accountCREATEUSER [bi_service] WITH PASSWORD ='${BI_SERVICE_PASSWORD}';
GRANTSELECTON SCHEMA::gold TO [bi_service];
Monitoring & Logging
-- Create audit log tableCREATE TABLE dbo.ETL_AuditLog (
LogID INTIDENTITY(1,1) PRIMARY KEY,
ProcedureName NVARCHAR(255),
LayerName NVARCHAR(50),
StartTime DATETIME2,
EndTime DATETIME2,
RowsProcessed INT,
Status NVARCHAR(50),
ErrorMessage NVARCHAR(MAX)
);
-- Example audit logging in proceduresCREATEPROCEDURE silver.usp_TransformSalesWithLogging
ASBEGINDECLARE@StartTime DATETIME2 = GETDATE();
DECLARE@RowCountINT;
BEGIN TRY
-- Transform logicINSERT INTO silver.Sales (...) SELECT ...;
SET@RowCount= @@ROWCOUNT;
-- Log successINSERT INTO dbo.ETL_AuditLog (ProcedureName, LayerName, StartTime, EndTime, RowsProcessed, Status)
VALUES ('usp_TransformSales', 'Silver', @StartTime, GETDATE(), @RowCount, 'Success');
END TRY
BEGIN CATCH
-- Log failureINSERT INTO dbo.ETL_AuditLog (ProcedureName, LayerName, StartTime, EndTime, Status, ErrorMessage)
VALUES ('usp_TransformSales', 'Silver', @StartTime, GETDATE(), 'Failed', ERROR_MESSAGE());
THROW;
END CATCH;
END;
This skill provides comprehensive guidance for implementing and extending the Retail ETL Medallion Pipeline with real-world business logic and production-ready patterns.
