Thapelo Tshiamo
Moloto

-- SCD Type 2: Customer Dimension with effective dating CREATE TABLE dim_customer ( customer_sk BIGINT IDENTITY(1,1) PRIMARY KEY, customer_bk VARCHAR(50) NOT NULL, customer_name VARCHAR(200), customer_segment VARCHAR(50), region VARCHAR(100), effective_from DATE NOT NULL, effective_to DATE DEFAULT '9999-12-31', is_current BIT DEFAULT 1, row_hash BINARY(32) -- SHA2 hash for change detection ); -- Clustered index: business key + effective date (range scans) CREATE CLUSTERED INDEX cidx_cust_bk_eff ON dim_customer (customer_bk, effective_from); -- Filtered non-clustered: only live rows (current-record lookups) CREATE NONCLUSTERED INDEX nidx_cust_current ON dim_customer (customer_bk) INCLUDE (customer_sk, customer_name, region) WHERE is_current = 1; -- SCD Type 2 MERGE: expire changed rows, insert new versions MERGE INTO dim_customer AS tgt USING ( SELECT s.customer_bk, s.customer_name, s.customer_segment, s.region, HASHBYTES('SHA2_256', COALESCE(s.customer_name,'') + COALESCE(s.customer_segment,'') + COALESCE(s.region,'')) AS row_hash FROM stg_customer s ) AS src ON tgt.customer_bk = src.customer_bk AND tgt.is_current = 1 WHEN MATCHED AND tgt.row_hash <> src.row_hash THEN UPDATE SET tgt.effective_to = CAST(GETDATE() AS DATE) - 1, tgt.is_current = 0 WHEN NOT MATCHED THEN INSERT (customer_bk, customer_name, customer_segment, region, effective_from, row_hash) VALUES (src.customer_bk, src.customer_name, src.customer_segment, src.region, CAST(GETDATE() AS DATE), src.row_hash);

-- Monthly partition scheme for incremental Power BI refresh CREATE PARTITION FUNCTION pf_monthly (DATE) AS RANGE RIGHT FOR VALUES ('2023-01-01','2023-02-01','2023-03-01', '2023-04-01','2023-05-01','2023-06-01','2023-07-01', '2023-08-01','2023-09-01','2023-10-01','2023-11-01','2023-12-01'); CREATE PARTITION SCHEME ps_monthly AS PARTITION pf_monthly ALL TO ([PRIMARY]); -- Fact table partitioned by sale_date CREATE TABLE fact_sales ( sale_sk BIGINT IDENTITY(1,1), sale_date DATE NOT NULL, customer_sk BIGINT NOT NULL, product_sk BIGINT NOT NULL, store_sk BIGINT NOT NULL, revenue DECIMAL(18,2), units_sold INT ) ON ps_monthly(sale_date); CREATE CLUSTERED COLUMNSTORE INDEX cci_fact_sales ON fact_sales ON ps_monthly(sale_date); -- Materialized view: monthly revenue pre-aggregated for DirectQuery CREATE MATERIALIZED VIEW mv_monthly_revenue WITH (DISTRIBUTION = HASH(store_sk)) AS SELECT DATEFROMPARTS(YEAR(f.sale_date), MONTH(f.sale_date), 1) AS month_start, d.region, p.category, SUM(f.revenue) AS total_revenue, SUM(f.units_sold) AS total_units, COUNT_BIG(DISTINCT f.customer_sk) AS unique_customers FROM fact_sales f JOIN dim_customer d ON f.customer_sk = d.customer_sk AND d.is_current = 1 JOIN dim_product p ON f.product_sk = p.product_sk GROUP BY DATEFROMPARTS(YEAR(f.sale_date),MONTH(f.sale_date),1), d.region, p.category;

# PySpark: write Gold layer partitioned by Policy_Year from pyspark.sql import functions as F gold_df = ( silver_claims_df .withColumn("Policy_Year", F.year("policy_start_date")) .withColumn("claim_amount", F.coalesce(F.col("claim_amount"), F.lit(0))) ) (gold_df.write .format("delta") .mode("overwrite") .option("overwriteSchema", "true") .partitionBy("Policy_Year") .save("abfss://gold@datalake.dfs.core.windows.net/insurance/claims") ) -- Z-Order: cluster Claim_ID + Policy_Number within each partition OPTIMIZE gold.insurance_claims ZORDER BY (Claim_ID, Policy_Number); -- Retain 7-day time-travel, purge stale files VACUUM gold.insurance_claims RETAIN 168 HOURS;

WITH claim_timeline AS ( SELECT c.customer_id, cl.claim_date, cl.claim_amount, cl.claim_type, DATEDIFF('day', LAG(cl.claim_date) OVER (PARTITION BY c.customer_id ORDER BY cl.claim_date), cl.claim_date ) AS days_since_prev_claim, SUM(cl.claim_amount) OVER ( PARTITION BY c.customer_id ORDER BY cl.claim_date ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW ) AS cumulative_exposure, DENSE_RANK() OVER ( PARTITION BY c.customer_id, YEAR(cl.claim_date) ORDER BY cl.claim_amount DESC ) AS claim_rank_ytd FROM fact_claims cl JOIN dim_customer c ON cl.customer_id = c.customer_id JOIN dim_policy p ON cl.policy_number = p.policy_number JOIN dim_product pr ON p.product_code = pr.product_code WHERE p.policy_year = YEAR(CURRENT_DATE()) - 1 ), risk_flags AS ( SELECT *, CASE WHEN days_since_prev_claim < 30 THEN 'HIGH_FREQUENCY' WHEN cumulative_exposure > 500000 THEN 'HIGH_EXPOSURE' WHEN claim_rank_ytd = 1 AND claim_amount > 100000 THEN 'LARGE_SINGLE_CLAIM' ELSE 'NORMAL' END AS risk_category FROM claim_timeline ) SELECT * FROM risk_flags WHERE risk_category <> 'NORMAL' ORDER BY cumulative_exposure DESC;

# zenith_migrator/cli.py import click, pyodbc from azure.storage.filedatalake import DataLakeServiceClient @click.command() @click.option("--table", required=True) @click.option("--batch", default=50000) @click.option("--dry-run", is_flag=True) def migrate(table, batch, dry_run): """Migrate a SQL Server table to ADLS with checksum audit.""" schema_diff = compare_schemas(get_source_schema(table), get_target_schema(table)) if schema_diff: click.echo(f"Schema mismatch: {schema_diff}", err=True) raise click.Abort() src = compute_checksum(table, source=True) click.echo(f"Source: {src['row_count']:,} rows | hash={src['col_hash']}") if not dry_run: bulk_migrate(table, batch_size=batch) tgt = compute_checksum(table, source=False) if src['row_count'] != tgt['row_count']: raise RuntimeError("DATA LOSS DETECTED") click.echo(f"Verified: {tgt['row_count']:,} rows | 0% data loss") def compute_checksum(table, source=True): query = f""" SELECT COUNT(*) AS row_count, CHECKSUM_AGG(CHECKSUM(*)) AS col_hash FROM [{table}] WITH (NOLOCK) """ with (get_src_conn() if source else get_tgt_conn()) as conn: r = conn.execute(query).fetchone() return {"row_count": r[0], "col_hash": r[1]}

-- Drop index before bulk load (enables minimal logging) IF EXISTS (SELECT 1 FROM sys.indexes WHERE object_id = OBJECT_ID('dbo.migration_staging') AND name = 'cidx_staging_pk') DROP INDEX cidx_staging_pk ON dbo.migration_staging; -- Bulk load with TABLOCK for minimal logging BULK INSERT dbo.migration_staging FROM '\\staging\export\orders_001.csv' WITH ( FIELDTERMINATOR = ',', ROWTERMINATOR = '\n', FIRSTROW = 2, BATCHSIZE = 50000, TABLOCK, CHECK_CONSTRAINTS, FIRE_TRIGGERS ); -- Rebuild clustered index after all batches complete CREATE CLUSTERED INDEX cidx_staging_pk ON dbo.migration_staging (order_id) WITH (SORT_IN_TEMPDB = ON, FILLFACTOR = 90, ONLINE = OFF); -- Non-clustered covering index for query patterns CREATE NONCLUSTERED INDEX nidx_staging_cust_date ON dbo.migration_staging (customer_id, order_date) INCLUDE (order_total, status) WITH (FILLFACTOR = 85, ONLINE = OFF); -- Row-count integrity guard DECLARE @src BIGINT = 1412847; DECLARE @tgt BIGINT = (SELECT COUNT_BIG(*) FROM dbo.migration_staging WITH(NOLOCK)); IF @src <> @tgt THROW 50001, 'DATA LOSS: row count mismatch post-migration.', 1; PRINT 'Migration audit PASSED — zero data loss confirmed.';

-- Partitioned tick table, composite cluster key CREATE TABLE market_ticks ( tick_id BIGINT IDENTITY(1,1), ticker VARCHAR(10) NOT NULL, trade_ts DATETIME2(6) NOT NULL, open_price DECIMAL(18,6), high_price DECIMAL(18,6), low_price DECIMAL(18,6), close_price DECIMAL(18,6), volume BIGINT, trade_date DATE NOT NULL ) ON ps_trade_daily(trade_date); -- Clustered composite: ticker first, then timestamp CREATE CLUSTERED INDEX cidx_ticks_ticker_ts ON market_ticks (ticker ASC, trade_ts ASC); -- Covering NCCI: OHLCV on the same key avoids key lookups CREATE NONCLUSTERED INDEX nidx_ticks_ohlcv ON market_ticks (ticker, trade_ts) INCLUDE (open_price, high_price, low_price, close_price, volume) WITH (FILLFACTOR = 80);

WITH price_base AS ( SELECT ticker, trade_ts, trade_date, close_price, volume, AVG(close_price) OVER ( PARTITION BY ticker ORDER BY trade_ts ROWS BETWEEN 4 PRECEDING AND CURRENT ROW ) AS sma_5, AVG(close_price) OVER ( PARTITION BY ticker ORDER BY trade_ts ROWS BETWEEN 29 PRECEDING AND CURRENT ROW ) AS sma_30, STDEVP(close_price) OVER ( PARTITION BY ticker ORDER BY trade_ts ROWS BETWEEN 29 PRECEDING AND CURRENT ROW ) AS volatility_30, SUM(close_price * volume) OVER ( PARTITION BY ticker, trade_date ORDER BY trade_ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW ) / NULLIF(SUM(volume) OVER ( PARTITION BY ticker, trade_date ORDER BY trade_ts ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW), 0 ) AS vwap, (close_price - LAG(close_price,1) OVER (PARTITION BY ticker ORDER BY trade_ts)) / NULLIF(LAG(close_price,1) OVER (PARTITION BY ticker ORDER BY trade_ts),0) AS pct_return FROM market_ticks WHERE trade_date >= DATEADD(day,-90,CAST(GETDATE() AS DATE)) ), signals AS ( SELECT *, CASE WHEN sma_5 > sma_30 AND LAG(sma_5) OVER (PARTITION BY ticker ORDER BY trade_ts) <= LAG(sma_30) OVER (PARTITION BY ticker ORDER BY trade_ts) THEN 'GOLDEN_CROSS' WHEN sma_5 < sma_30 AND LAG(sma_5) OVER (PARTITION BY ticker ORDER BY trade_ts) >= LAG(sma_30) OVER (PARTITION BY ticker ORDER BY trade_ts) THEN 'DEATH_CROSS' END AS signal FROM price_base ) SELECT ticker, trade_ts, close_price, ROUND(sma_5,4) AS sma_5, ROUND(sma_30,4) AS sma_30, ROUND(volatility_30,6) AS volatility, ROUND(vwap,4) AS vwap, ROUND(pct_return*100,4) AS pct_return, signal FROM signals WHERE signal IS NOT NULL ORDER BY trade_ts DESC;