undefined

undefined

pip install pandas sqlalchemy pyodbc pyspark matplotlib

-- 01_Schema_Setup.sql
-- Create the data warehouse database
CREATE DATABASE EnterpriseDataWarehouse;
GO

USE EnterpriseDataWarehouse;
GO

-- Dimension: Customers
CREATE TABLE dim_Customers (
    CustomerID INT PRIMARY KEY,
    CustomerName NVARCHAR(100),
    Email NVARCHAR(100),
    Region NVARCHAR(50),
    RegistrationDate DATE
);

-- Dimension: Products
CREATE TABLE dim_Products (
    ProductID INT PRIMARY KEY,
    ProductName NVARCHAR(100),
    Category NVARCHAR(50),
    UnitPrice DECIMAL(10, 2)
);

-- Fact: Sales
CREATE TABLE fact_Sales (
    SaleID INT PRIMARY KEY,
    CustomerID INT FOREIGN KEY REFERENCES dim_Customers(CustomerID),
    ProductID INT FOREIGN KEY REFERENCES dim_Products(ProductID),
    Quantity INT,
    SaleDate DATE,
    TotalAmount DECIMAL(10, 2)
);

-- Business Intelligence View: Revenue by Product
CREATE VIEW vw_RevenueByProduct AS
SELECT 
    p.ProductName,
    p.Category,
    SUM(s.TotalAmount) AS TotalRevenue,
    SUM(s.Quantity) AS TotalQuantity
FROM fact_Sales s
INNER JOIN dim_Products p ON s.ProductID = p.ProductID
GROUP BY p.ProductName, p.Category;

-- Business Intelligence View: Customer Lifetime Value
CREATE VIEW vw_CustomerLTV AS
SELECT 
    c.CustomerID,
    c.CustomerName,
    c.Region,
    COUNT(s.SaleID) AS TotalPurchases,
    SUM(s.TotalAmount) AS LifetimeValue
FROM dim_Customers c
LEFT JOIN fact_Sales s ON c.CustomerID = s.CustomerID
GROUP BY c.CustomerID, c.CustomerName, c.Region;

<!-- Key SSIS Components -->
<!-- Data Flow Task: Load dim_Customers -->
- Flat File Source (Customers.csv)
- Data Conversion (handle Unicode, trim strings)
- Derived Column (add audit columns)
- OLEDB Destination (dim_Customers)

<!-- Data Flow Task: Load dim_Products -->
- Flat File Source (Products.csv)
- Data Conversion (decimal precision for prices)
- OLEDB Destination (dim_Products)

<!-- Data Flow Task: Load fact_Sales -->
- Flat File Source (Sales.csv)
- Lookup Transformation (validate CustomerID, ProductID)
- Derived Column (calculate TotalAmount = Quantity * UnitPrice)
- OLEDB Destination (fact_Sales)

-- Create error logging table
CREATE TABLE ETL_ErrorLog (
    ErrorID INT IDENTITY(1,1) PRIMARY KEY,
    PackageName NVARCHAR(100),
    TaskName NVARCHAR(100),
    ErrorDescription NVARCHAR(MAX),
    ErrorDate DATETIME DEFAULT GETDATE()
);

undefined

# Check 1: Null values in critical columns
query_nulls = """
SELECT 
    'dim_Customers' AS TableName,
    SUM(CASE WHEN CustomerName IS NULL THEN 1 ELSE 0 END) AS NullCustomerName,
    SUM(CASE WHEN Email IS NULL THEN 1 ELSE 0 END) AS NullEmail
FROM dim_Customers
UNION ALL
SELECT 
    'fact_Sales',
    SUM(CASE WHEN CustomerID IS NULL THEN 1 ELSE 0 END),
    SUM(CASE WHEN ProductID IS NULL THEN 1 ELSE 0 END)
FROM fact_Sales
"""
df_nulls = pd.read_sql(query_nulls, engine)
print("Null Value Audit:")
print(df_nulls)

# Check 2: Orphaned records (referential integrity)
query_orphans = """
SELECT COUNT(*) AS OrphanedSales
FROM fact_Sales s
WHERE NOT EXISTS (SELECT 1 FROM dim_Customers c WHERE c.CustomerID = s.CustomerID)
   OR NOT EXISTS (SELECT 1 FROM dim_Products p WHERE p.ProductID = s.ProductID)
"""
df_orphans = pd.read_sql(query_orphans, engine)
print("\nOrphaned Records:")
print(df_orphans)

# Check 3: Revenue distribution
query_revenue = "SELECT * FROM vw_RevenueByProduct ORDER BY TotalRevenue DESC"
df_revenue = pd.read_sql(query_revenue, engine)

# Visualization
plt.figure(figsize=(10, 6))
plt.bar(df_revenue['ProductName'][:10], df_revenue['TotalRevenue'][:10])
plt.xlabel('Product')
plt.ylabel('Total Revenue')
plt.title('Top 10 Products by Revenue')
plt.xticks(rotation=45, ha='right')
plt.tight_layout()
plt.savefig('revenue_analysis.png')
print("\nRevenue chart saved to revenue_analysis.png")

undefined

undefined

# Load customer LTV data
df = pd.read_sql("SELECT * FROM vw_CustomerLTV", engine)

# RFM-style segmentation
df['Segment'] = pd.cut(
    df['LifetimeValue'],
    bins=[0, 1000, 5000, float('inf')],
    labels=['Bronze', 'Silver', 'Gold']
)

# Aggregate by segment
segment_summary = df.groupby('Segment').agg({
    'CustomerID': 'count',
    'LifetimeValue': 'sum',
    'TotalPurchases': 'mean'
}).reset_index()

print(segment_summary)
return segment_summary

undefined

undefined

undefined

undefined

undefined

-- Create staging table
CREATE TABLE stg_Sales (
    SaleID INT,
    CustomerID INT,
    ProductID INT,
    Quantity INT,
    SaleDate DATE,
    TotalAmount DECIMAL(10, 2),
    LoadDate DATETIME DEFAULT GETDATE()
);

-- Merge statement for incremental load
MERGE INTO fact_Sales AS target
USING stg_Sales AS source
ON target.SaleID = source.SaleID
WHEN MATCHED THEN
    UPDATE SET
        Quantity = source.Quantity,
        TotalAmount = source.TotalAmount
WHEN NOT MATCHED THEN
    INSERT (SaleID, CustomerID, ProductID, Quantity, SaleDate, TotalAmount)
    VALUES (source.SaleID, source.CustomerID, source.ProductID, 
            source.Quantity, source.SaleDate, source.TotalAmount);

undefined

cmd = [
    "dtexec",
    "/FILE", package_path,
    "/REPORTING", "E"
]

result = subprocess.run(cmd, capture_output=True, text=True)

log_file = f"etl_log_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
with open(log_file, 'w') as f:
    f.write(result.stdout)
    f.write(result.stderr)

return result.returncode == 0

undefined

Error: "Cannot acquire connection to SQL Server"
Solution:
1. Verify SQL Server service is running
2. Check Windows Authentication vs SQL Authentication
3. Update connection string in Connection Manager
4. Enable TCP/IP protocol in SQL Server Configuration Manager

Error: "Cannot convert between unicode and non-unicode string data types"
Solution in SSIS:
1. Add Data Conversion task
2. Convert DT_STR to DT_WSTR for Unicode columns
3. Set CodePage to 1252 (Windows Latin 1) in Flat File Connection

undefined

undefined

undefined

undefined

undefined

undefined

undefined

This enterprise pipeline provides a complete solution for data warehousing, ETL automation, and big data analytics using industry-standard Microsoft and Apache technologies.