Microsoft Fabric has rapidly become a cornerstone for unified analytics workloads across US enterprises, combining data lakehouse capabilities with real-time analytics in a single SaaS platform. However, as organizations scale their Fabric deployments beyond initial proof-of-concepts, query performance bottlenecks emerge that can cripple BI dashboards, slow analytical workflows, and inflate compute costs. The performance challenges are particularly acute given Fabric's multi-engine architecture spanning SQL Analytics Endpoint, Warehouse, and Lakehouse compute layers.

Senior data engineering teams consistently report three critical pain points: inconsistent dashboard latency across different Fabric engines, unpredictable query performance when scaling from gigabyte to terabyte datasets, and difficulty optimizing cross-engine workloads that span both structured warehouses and semi-structured lakehouse data. This playbook provides battle-tested optimization tactics specifically designed for Fabric's unique architecture, with runnable code examples and clear guidance on when each approach delivers maximum performance impact.

Performance Yardsticks

• Dashboard Query Latency · Good Performance: <3s p95 response · Needs Attention: 3-10s avg response · Critical Issue: >10s for simple queries

• Lakehouse Scan Performance · Good Performance: <5s for 10GB Delta tables · Needs Attention: 5-30s on partitioned data · Critical Issue: >30s on optimized tables

• Warehouse Complex JOIN Latency · Good Performance: <15s for 5-table JOINs · Needs Attention: 15-60s with proper indexing · Critical Issue: >60s with star schema

• Real-time Analytics Throughput · Good Performance: >1000 queries/hr sustained · Needs Attention: 500-1000 queries/hr peak · Critical Issue: <500 queries/hr degradation

• Cross-engine Query Performance · Good Performance: <20s Lakehouse→Warehouse · Needs Attention: 20-120s with shortcuts · Critical Issue: >120s for federated queries

• Data Loading Throughput · Good Performance: >100MB/s per CU sustained · Needs Attention: 50-100MB/s during peak · Critical Issue: <50MB/s with contention

• Concurrent User Scalability · Good Performance: No SLA degradation <100 users · Needs Attention: Slight latency increase 100-300 · Critical Issue: Query timeouts >300 users

‍

Workload Taxonomy

• BI Dashboards · Characteristics: High-frequency analytical queries with predictable access patterns. Typically 5-20 visualizations per dashboard pulling from star/snowflake schemas with <1M fact table rows scanned per query. · Performance Requirements: Sub-3s p95 latency, 100+ concurrent users, consistent performance across peak hours · Common Bottlenecks: Inefficient partition elimination, missing columnstore compression, cross-engine query federation overhead

• Ad-hoc Analytics · Characteristics: Exploratory data science and business analyst queries with unpredictable JOIN patterns. Often involves complex window functions, CTEs, and multi-table aggregations across 10GB+ datasets. · Performance Requirements: <30s for complex analysis, query result caching, interactive exploration experience · Common Bottlenecks: Full table scans on Delta tables, memory spill in complex JOINs, suboptimal predicate pushdown

• ETL/Streaming · Characteristics: High-volume data processing pipelines with both batch and near real-time requirements. Includes Dataflow Gen2 transformations, Data Pipeline orchestration, and streaming analytics. · Performance Requirements: >1GB/min throughput, <5min end-to-end latency for streaming, parallel processing capability · Common Bottlenecks: Notebook memory limits, inefficient Delta table writes, streaming micro-batch optimization, cross-workspace data movement

‍

BI Dashboards Optimization Tactics

Implement Delta Table Z-ORDER for dashboard access patterns

When your Power BI dashboards consistently query specific dimensional combinations (like region + product category + time period), Z-ORDER clustering dramatically improves query performance by co-locating related data within the same data files. You'll see substantially faster dashboard load times when Z-ORDER columns match your most common filter combinations.

The key insight here is understanding your dashboard's query patterns before implementing Z-ORDER. Most enterprise dashboards follow predictable access patterns where users filter by date ranges, geographic regions, or business units. Here's how to implement Z-ORDER optimization for a typical sales dashboard scenario:

-- Optimize sales fact table for common dashboard filters
OPTIMIZE sales_fact_delta
ZORDER BY (region_id, product_category, sale_date);

-- Verify Z-ORDER effectiveness with file statistics
DESCRIBE DETAIL sales_fact_delta;

-- Optimize sales fact table for common dashboard filters
OPTIMIZE sales_fact_delta
ZORDER BY (region_id, product_category, sale_date);

-- Verify Z-ORDER effectiveness with file statistics
DESCRIBE DETAIL sales_fact_delta;

-- Optimize sales fact table for common dashboard filters
OPTIMIZE sales_fact_delta
ZORDER BY (region_id, product_category, sale_date);

-- Verify Z-ORDER effectiveness with file statistics
DESCRIBE DETAIL sales_fact_delta;

-- Optimize sales fact table for common dashboard filters
OPTIMIZE sales_fact_delta
ZORDER BY (region_id, product_category, sale_date);

-- Verify Z-ORDER effectiveness with file statistics
DESCRIBE DETAIL sales_fact_delta;

What makes this particularly effective is that Z-ORDER works at the Parquet file level, reducing the number of files Fabric needs to scan during query execution. When users filter dashboards by "West Region + Electronics + Last 30 Days", Fabric can skip entire files that don't contain relevant data combinations.

Alternative approaches include: V-ORDER clustering for write-heavy scenarios (superior for ETL pipelines but less dashboard optimization), traditional table partitioning by date (simpler implementation but reduced flexibility for multi-dimensional filters), Microsoft Fabric API automation for dynamic optimization, or leveraging e6data as a complementary lakehouse compute engine that automatically optimizes data layout without manual clustering commands while delivering sub-second dashboard latency through its decentralized architecture.

Configure Power BI Aggregations with Microsoft Fabric Warehouse

Dashboard performance bottlenecks often stem from Power BI repeatedly calculating the same aggregations across millions of fact table rows. Power BI aggregations combined with Fabric Warehouse materialized views eliminate this computational overhead by pre-calculating common dashboard metrics.

-- Create warehouse materialized view for monthly sales aggregations
CREATE MATERIALIZED VIEW monthly_sales_agg AS
SELECT 
    region_id,
    product_category,
    YEAR(sale_date) as sale_year,
    MONTH(sale_date) as sale_month,
    SUM(revenue) as total_revenue,
    COUNT(*) as transaction_count,
    AVG(revenue) as avg_transaction_value
FROM sales_fact
WHERE sale_date >= DATEADD(YEAR, -2, GETDATE())
GROUP BY region_id, product_category, YEAR(sale_date), MONTH(sale_date);

-- Configure automatic refresh schedule
ALTER MATERIALIZED VIEW monthly_sales_agg SET (AUTO_REFRESH = ON);

-- Create warehouse materialized view for monthly sales aggregations
CREATE MATERIALIZED VIEW monthly_sales_agg AS
SELECT 
    region_id,
    product_category,
    YEAR(sale_date) as sale_year,
    MONTH(sale_date) as sale_month,
    SUM(revenue) as total_revenue,
    COUNT(*) as transaction_count,
    AVG(revenue) as avg_transaction_value
FROM sales_fact
WHERE sale_date >= DATEADD(YEAR, -2, GETDATE())
GROUP BY region_id, product_category, YEAR(sale_date), MONTH(sale_date);

-- Configure automatic refresh schedule
ALTER MATERIALIZED VIEW monthly_sales_agg SET (AUTO_REFRESH = ON);

-- Create warehouse materialized view for monthly sales aggregations
CREATE MATERIALIZED VIEW monthly_sales_agg AS
SELECT 
    region_id,
    product_category,
    YEAR(sale_date) as sale_year,
    MONTH(sale_date) as sale_month,
    SUM(revenue) as total_revenue,
    COUNT(*) as transaction_count,
    AVG(revenue) as avg_transaction_value
FROM sales_fact
WHERE sale_date >= DATEADD(YEAR, -2, GETDATE())
GROUP BY region_id, product_category, YEAR(sale_date), MONTH(sale_date);

-- Configure automatic refresh schedule
ALTER MATERIALIZED VIEW monthly_sales_agg SET (AUTO_REFRESH = ON);

-- Create warehouse materialized view for monthly sales aggregations
CREATE MATERIALIZED VIEW monthly_sales_agg AS
SELECT 
    region_id,
    product_category,
    YEAR(sale_date) as sale_year,
    MONTH(sale_date) as sale_month,
    SUM(revenue) as total_revenue,
    COUNT(*) as transaction_count,
    AVG(revenue) as avg_transaction_value
FROM sales_fact
WHERE sale_date >= DATEADD(YEAR, -2, GETDATE())
GROUP BY region_id, product_category, YEAR(sale_date), MONTH(sale_date);

-- Configure automatic refresh schedule
ALTER MATERIALIZED VIEW monthly_sales_agg SET (AUTO_REFRESH = ON);

Here's where it gets interesting: Power BI's query engine will automatically detect when dashboard visuals can be satisfied by your materialized view instead of scanning the underlying fact table. This happens transparently to end users, but the performance impact is dramatic for dashboards that aggregate data across time periods or business dimensions.

Alternative approaches include: Incremental refresh for large semantic models (reduces data transfer but doesn't eliminate aggregation compute), or composite models with mixed storage modes (more complex setup but allows hybrid cloud/on-premises scenarios).

Optimize Cross-engine Queries with Microsoft OneLake Shortcuts

When dashboards pull data from both Lakehouse Delta tables and Warehouse structured tables, Fabric Shortcuts eliminate expensive cross-engine data movement by creating unified virtual views of your data landscape. This prevents the performance penalty of federating queries across different Fabric compute engines during dashboard refresh.

What typically happens without shortcuts is that Power BI forces Fabric to orchestrate complex cross-engine queries where warehouse queries need to JOIN with lakehouse data. This coordination overhead can significantly increase dashboard refresh times. Here's how to implement shortcuts for optimal dashboard performance:

-- Create lakehouse shortcut to warehouse dimension tables
-- Execute in Lakehouse SQL endpoint
CREATE SHORTCUT [warehouse_dims]
TO 'sql-endpoint://your-workspace.datawarehouse.fabric.microsoft.com/your-warehouse/Schemas/dbo/Tables/dim_product'
WITH (TYPE = 'SQL_ENDPOINT');

-- Now query unified data without cross-engine overhead
SELECT 
    f.sale_date,
    d.product_name,
    d.category,
    SUM(f.revenue) as daily_revenue
FROM sales_fact_delta f
INNER JOIN warehouse_dims.dim_product d ON f.product_id = d.product_id
WHERE f.sale_date >= DATEADD(DAY, -30, GETDATE())
GROUP BY f.sale_date, d.product_name, d.category;

-- Materialized shortcuts to eliminate cross-engine overhead
CREATE SHORTCUT [lakehouse_warehouse_bridge]
TO 'sql-endpoint://workspace.datawarehouse.fabric.microsoft.com/warehouse/Schemas/dbo/Tables/dim_customer'
WITH (TYPE = 'SQL_ENDPOINT');

-- Pre-aggregate across engines to minimize query-time joins
CREATE MATERIALIZED VIEW cross_engine_agg AS
SELECT 
    lh.customer_id,
    lh.transaction_date,
    SUM(lh.revenue) as lakehouse_revenue,
    wh.customer_segment,
    wh.lifetime_value
FROM lakehouse.sales_fact lh
INNER JOIN warehouse.customer_master wh ON lh.customer_id = wh.customer_id
WHERE lh.transaction_date >= DATEADD(MONTH, -12, GETDATE())
GROUP BY lh.customer_id, lh.transaction_date, wh.customer_segment, wh.lifetime_value;
# Smart data movement patterns to optimize cross-engine performance
def optimize_cross_engine_performance():
    """Minimize data movement between Fabric engines"""
    
    # Strategy 1: Co-locate frequently joined data
    def colocate_related_data():
        # Move dimension tables to lakehouse for better join performance
        customer_dim = spark.sql("SELECT * FROM warehouse.customer_master")
        customer_dim.write.format("delta").mode("overwrite").saveAsTable("lakehouse.customer_dim_copy")
        
        # Create bidirectional shortcuts for critical tables
        shortcuts_config = [
            {'source': 'warehouse.customer_master', 'target': 'lakehouse.customer_shortcut'},
            {'source': 'lakehouse.sales_fact', 'target': 'warehouse.sales_shortcut'}
        ]
        
        return shortcuts_config
    
    # Strategy 2: Implement query result caching across engines
    def cache_cross_engine_results():
        cross_engine_query = """
        SELECT 
            w.customer_segment,
            l.product_category,
            SUM(l.revenue) as total_revenue,
            COUNT(*) as transaction_count
        FROM lakehouse.sales_fact l
        INNER JOIN warehouse.customer_master w ON l.customer_id = w.customer_id
        WHERE l.sale_date >= CURRENT_DATE - INTERVAL 30 DAYS
        GROUP BY w.customer_segment, l.product_category
        """
        
        result_df = spark.sql(cross_engine_query)
        result_df.cache()
        result_df.write.format("delta").mode("overwrite").saveAsTable("cache.cross_engine_summary")
        
        return result_df

optimize_cross_engine_performance()

-- Create lakehouse shortcut to warehouse dimension tables
-- Execute in Lakehouse SQL endpoint
CREATE SHORTCUT [warehouse_dims]
TO 'sql-endpoint://your-workspace.datawarehouse.fabric.microsoft.com/your-warehouse/Schemas/dbo/Tables/dim_product'
WITH (TYPE = 'SQL_ENDPOINT');

-- Now query unified data without cross-engine overhead
SELECT 
    f.sale_date,
    d.product_name,
    d.category,
    SUM(f.revenue) as daily_revenue
FROM sales_fact_delta f
INNER JOIN warehouse_dims.dim_product d ON f.product_id = d.product_id
WHERE f.sale_date >= DATEADD(DAY, -30, GETDATE())
GROUP BY f.sale_date, d.product_name, d.category;

-- Materialized shortcuts to eliminate cross-engine overhead
CREATE SHORTCUT [lakehouse_warehouse_bridge]
TO 'sql-endpoint://workspace.datawarehouse.fabric.microsoft.com/warehouse/Schemas/dbo/Tables/dim_customer'
WITH (TYPE = 'SQL_ENDPOINT');

-- Pre-aggregate across engines to minimize query-time joins
CREATE MATERIALIZED VIEW cross_engine_agg AS
SELECT 
    lh.customer_id,
    lh.transaction_date,
    SUM(lh.revenue) as lakehouse_revenue,
    wh.customer_segment,
    wh.lifetime_value
FROM lakehouse.sales_fact lh
INNER JOIN warehouse.customer_master wh ON lh.customer_id = wh.customer_id
WHERE lh.transaction_date >= DATEADD(MONTH, -12, GETDATE())
GROUP BY lh.customer_id, lh.transaction_date, wh.customer_segment, wh.lifetime_value;
# Smart data movement patterns to optimize cross-engine performance
def optimize_cross_engine_performance():
    """Minimize data movement between Fabric engines"""
    
    # Strategy 1: Co-locate frequently joined data
    def colocate_related_data():
        # Move dimension tables to lakehouse for better join performance
        customer_dim = spark.sql("SELECT * FROM warehouse.customer_master")
        customer_dim.write.format("delta").mode("overwrite").saveAsTable("lakehouse.customer_dim_copy")
        
        # Create bidirectional shortcuts for critical tables
        shortcuts_config = [
            {'source': 'warehouse.customer_master', 'target': 'lakehouse.customer_shortcut'},
            {'source': 'lakehouse.sales_fact', 'target': 'warehouse.sales_shortcut'}
        ]
        
        return shortcuts_config
    
    # Strategy 2: Implement query result caching across engines
    def cache_cross_engine_results():
        cross_engine_query = """
        SELECT 
            w.customer_segment,
            l.product_category,
            SUM(l.revenue) as total_revenue,
            COUNT(*) as transaction_count
        FROM lakehouse.sales_fact l
        INNER JOIN warehouse.customer_master w ON l.customer_id = w.customer_id
        WHERE l.sale_date >= CURRENT_DATE - INTERVAL 30 DAYS
        GROUP BY w.customer_segment, l.product_category
        """
        
        result_df = spark.sql(cross_engine_query)
        result_df.cache()
        result_df.write.format("delta").mode("overwrite").saveAsTable("cache.cross_engine_summary")
        
        return result_df

optimize_cross_engine_performance()

-- Create lakehouse shortcut to warehouse dimension tables
-- Execute in Lakehouse SQL endpoint
CREATE SHORTCUT [warehouse_dims]
TO 'sql-endpoint://your-workspace.datawarehouse.fabric.microsoft.com/your-warehouse/Schemas/dbo/Tables/dim_product'
WITH (TYPE = 'SQL_ENDPOINT');

-- Now query unified data without cross-engine overhead
SELECT 
    f.sale_date,
    d.product_name,
    d.category,
    SUM(f.revenue) as daily_revenue
FROM sales_fact_delta f
INNER JOIN warehouse_dims.dim_product d ON f.product_id = d.product_id
WHERE f.sale_date >= DATEADD(DAY, -30, GETDATE())
GROUP BY f.sale_date, d.product_name, d.category;

-- Materialized shortcuts to eliminate cross-engine overhead
CREATE SHORTCUT [lakehouse_warehouse_bridge]
TO 'sql-endpoint://workspace.datawarehouse.fabric.microsoft.com/warehouse/Schemas/dbo/Tables/dim_customer'
WITH (TYPE = 'SQL_ENDPOINT');

-- Pre-aggregate across engines to minimize query-time joins
CREATE MATERIALIZED VIEW cross_engine_agg AS
SELECT 
    lh.customer_id,
    lh.transaction_date,
    SUM(lh.revenue) as lakehouse_revenue,
    wh.customer_segment,
    wh.lifetime_value
FROM lakehouse.sales_fact lh
INNER JOIN warehouse.customer_master wh ON lh.customer_id = wh.customer_id
WHERE lh.transaction_date >= DATEADD(MONTH, -12, GETDATE())
GROUP BY lh.customer_id, lh.transaction_date, wh.customer_segment, wh.lifetime_value;
# Smart data movement patterns to optimize cross-engine performance
def optimize_cross_engine_performance():
    """Minimize data movement between Fabric engines"""
    
    # Strategy 1: Co-locate frequently joined data
    def colocate_related_data():
        # Move dimension tables to lakehouse for better join performance
        customer_dim = spark.sql("SELECT * FROM warehouse.customer_master")
        customer_dim.write.format("delta").mode("overwrite").saveAsTable("lakehouse.customer_dim_copy")
        
        # Create bidirectional shortcuts for critical tables
        shortcuts_config = [
            {'source': 'warehouse.customer_master', 'target': 'lakehouse.customer_shortcut'},
            {'source': 'lakehouse.sales_fact', 'target': 'warehouse.sales_shortcut'}
        ]
        
        return shortcuts_config
    
    # Strategy 2: Implement query result caching across engines
    def cache_cross_engine_results():
        cross_engine_query = """
        SELECT 
            w.customer_segment,
            l.product_category,
            SUM(l.revenue) as total_revenue,
            COUNT(*) as transaction_count
        FROM lakehouse.sales_fact l
        INNER JOIN warehouse.customer_master w ON l.customer_id = w.customer_id
        WHERE l.sale_date >= CURRENT_DATE - INTERVAL 30 DAYS
        GROUP BY w.customer_segment, l.product_category
        """
        
        result_df = spark.sql(cross_engine_query)
        result_df.cache()
        result_df.write.format("delta").mode("overwrite").saveAsTable("cache.cross_engine_summary")
        
        return result_df

optimize_cross_engine_performance()

-- Create lakehouse shortcut to warehouse dimension tables
-- Execute in Lakehouse SQL endpoint
CREATE SHORTCUT [warehouse_dims]
TO 'sql-endpoint://your-workspace.datawarehouse.fabric.microsoft.com/your-warehouse/Schemas/dbo/Tables/dim_product'
WITH (TYPE = 'SQL_ENDPOINT');

-- Now query unified data without cross-engine overhead
SELECT 
    f.sale_date,
    d.product_name,
    d.category,
    SUM(f.revenue) as daily_revenue
FROM sales_fact_delta f
INNER JOIN warehouse_dims.dim_product d ON f.product_id = d.product_id
WHERE f.sale_date >= DATEADD(DAY, -30, GETDATE())
GROUP BY f.sale_date, d.product_name, d.category;

-- Materialized shortcuts to eliminate cross-engine overhead
CREATE SHORTCUT [lakehouse_warehouse_bridge]
TO 'sql-endpoint://workspace.datawarehouse.fabric.microsoft.com/warehouse/Schemas/dbo/Tables/dim_customer'
WITH (TYPE = 'SQL_ENDPOINT');

-- Pre-aggregate across engines to minimize query-time joins
CREATE MATERIALIZED VIEW cross_engine_agg AS
SELECT 
    lh.customer_id,
    lh.transaction_date,
    SUM(lh.revenue) as lakehouse_revenue,
    wh.customer_segment,
    wh.lifetime_value
FROM lakehouse.sales_fact lh
INNER JOIN warehouse.customer_master wh ON lh.customer_id = wh.customer_id
WHERE lh.transaction_date >= DATEADD(MONTH, -12, GETDATE())
GROUP BY lh.customer_id, lh.transaction_date, wh.customer_segment, wh.lifetime_value;
# Smart data movement patterns to optimize cross-engine performance
def optimize_cross_engine_performance():
    """Minimize data movement between Fabric engines"""
    
    # Strategy 1: Co-locate frequently joined data
    def colocate_related_data():
        # Move dimension tables to lakehouse for better join performance
        customer_dim = spark.sql("SELECT * FROM warehouse.customer_master")
        customer_dim.write.format("delta").mode("overwrite").saveAsTable("lakehouse.customer_dim_copy")
        
        # Create bidirectional shortcuts for critical tables
        shortcuts_config = [
            {'source': 'warehouse.customer_master', 'target': 'lakehouse.customer_shortcut'},
            {'source': 'lakehouse.sales_fact', 'target': 'warehouse.sales_shortcut'}
        ]
        
        return shortcuts_config
    
    # Strategy 2: Implement query result caching across engines
    def cache_cross_engine_results():
        cross_engine_query = """
        SELECT 
            w.customer_segment,
            l.product_category,
            SUM(l.revenue) as total_revenue,
            COUNT(*) as transaction_count
        FROM lakehouse.sales_fact l
        INNER JOIN warehouse.customer_master w ON l.customer_id = w.customer_id
        WHERE l.sale_date >= CURRENT_DATE - INTERVAL 30 DAYS
        GROUP BY w.customer_segment, l.product_category
        """
        
        result_df = spark.sql(cross_engine_query)
        result_df.cache()
        result_df.write.format("delta").mode("overwrite").saveAsTable("cache.cross_engine_summary")
        
        return result_df

optimize_cross_engine_performance()

The key insight here is that shortcuts create a logical data mesh where your Power BI semantic model can treat lakehouse and warehouse data as if they exist in the same storage layer. This eliminates the query coordination overhead that typically slows down cross-engine dashboard queries.

Alternative approaches include: Data pipeline replication to consolidate data in single engine (increases storage costs and data freshness lag), DirectQuery optimization with query reduction techniques (reduces memory usage but maintains query latency), or e6data's lakehouse query engine that handles both structured and unstructured data.

Implement Smart Partitioning for time-series dashboards

Time-series dashboards consistently exhibit predictable access patterns where users focus on recent data (last 30-90 days) while occasionally drilling into historical trends. Fabric table partitioning by date, combined with proper partition elimination, ensures that dashboard queries only scan relevant data partitions.

-- Create partitioned warehouse table for time-series dashboard data
CREATE TABLE sales_fact_partitioned (
    sale_id BIGINT,
    customer_id INT,
    product_id INT,
    sale_date DATE,
    revenue DECIMAL(10,2),
    quantity INT
)
WITH (
    CLUSTERED COLUMNSTORE INDEX,
    PARTITION (sale_date 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',
        '2024-01-01', '2024-02-01', '2024-03-01', '2024-04-01'
    ))
);

-- For Delta tables in Lakehouse, use Hive-style partitioning
CREATE TABLE sales_fact_delta (
    sale_id BIGINT,
    customer_id INT,
    product_id INT,
    revenue DECIMAL(10,2),
    quantity INT,
    sale_date DATE
) 
USING DELTA
PARTITIONED BY (YEAR(sale_date), MONTH(sale_date));

-- Create partitioned warehouse table for time-series dashboard data
CREATE TABLE sales_fact_partitioned (
    sale_id BIGINT,
    customer_id INT,
    product_id INT,
    sale_date DATE,
    revenue DECIMAL(10,2),
    quantity INT
)
WITH (
    CLUSTERED COLUMNSTORE INDEX,
    PARTITION (sale_date 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',
        '2024-01-01', '2024-02-01', '2024-03-01', '2024-04-01'
    ))
);

-- For Delta tables in Lakehouse, use Hive-style partitioning
CREATE TABLE sales_fact_delta (
    sale_id BIGINT,
    customer_id INT,
    product_id INT,
    revenue DECIMAL(10,2),
    quantity INT,
    sale_date DATE
) 
USING DELTA
PARTITIONED BY (YEAR(sale_date), MONTH(sale_date));

-- Create partitioned warehouse table for time-series dashboard data
CREATE TABLE sales_fact_partitioned (
    sale_id BIGINT,
    customer_id INT,
    product_id INT,
    sale_date DATE,
    revenue DECIMAL(10,2),
    quantity INT
)
WITH (
    CLUSTERED COLUMNSTORE INDEX,
    PARTITION (sale_date 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',
        '2024-01-01', '2024-02-01', '2024-03-01', '2024-04-01'
    ))
);

-- For Delta tables in Lakehouse, use Hive-style partitioning
CREATE TABLE sales_fact_delta (
    sale_id BIGINT,
    customer_id INT,
    product_id INT,
    revenue DECIMAL(10,2),
    quantity INT,
    sale_date DATE
) 
USING DELTA
PARTITIONED BY (YEAR(sale_date), MONTH(sale_date));

-- Create partitioned warehouse table for time-series dashboard data
CREATE TABLE sales_fact_partitioned (
    sale_id BIGINT,
    customer_id INT,
    product_id INT,
    sale_date DATE,
    revenue DECIMAL(10,2),
    quantity INT
)
WITH (
    CLUSTERED COLUMNSTORE INDEX,
    PARTITION (sale_date 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',
        '2024-01-01', '2024-02-01', '2024-03-01', '2024-04-01'
    ))
);

-- For Delta tables in Lakehouse, use Hive-style partitioning
CREATE TABLE sales_fact_delta (
    sale_id BIGINT,
    customer_id INT,
    product_id INT,
    revenue DECIMAL(10,2),
    quantity INT,
    sale_date DATE
) 
USING DELTA
PARTITIONED BY (YEAR(sale_date), MONTH(sale_date));

Alternative approaches include: columnstore index optimization without partitioning (better for ad-hoc queries but less dashboard optimization), or table replication for small dimension tables (improves JOIN performance but increases storage).

Configure Fabric Capacity Auto-scaling for peak dashboard hours

Dashboard performance often degrades during business hours when hundreds of users simultaneously refresh their reports. Fabric Capacity auto-scaling prevents compute resource contention by automatically scaling compute units (CUs) based on query queue depth and CPU utilization patterns.

Here's what happens next: Fabric monitors your workspace's compute demand and automatically adds CUs when query queue times exceed your defined thresholds.

The key insight here is that auto-scaling prevents the query queueing that typically occurs when dashboard refresh jobs compete for limited compute resources. Instead of users experiencing timeouts or extended delays, auto-scaling maintains consistent dashboard response times.

Alternative approaches include: manual capacity scaling based on predictable usage patterns (more cost control but requires operational overhead), workload isolation to separate dashboard queries from ETL workloads (better resource allocation but more complex configuration), or e6data's per-vCPU scaling that eliminates capacity planning complexity while providing predictable costs and instant performance scaling without cluster management overhead.

Implement Direct Lake Cache Pre-warming for Optimal Dashboard Performance

Power BI dashboards using Direct Lake mode can experience significant performance degradation when the cache is cold, forcing fallback to DirectQuery mode which introduces substantial latency.

Direct Lake cache pre-warming ensures frequently accessed data remains memory-resident, delivering import-mode performance with real-time data freshness. When users access dashboards during peak business hours, data is already memory-resident, eliminating the cold-start latency that typically degrades user experience.

Alternative approaches include: incremental refresh policies for large semantic models (reduces refresh overhead but doesn't address cache warming), or composite models with strategic import/DirectQuery partitioning (more complex but handles mixed requirements).

Implement Direct Lake Fallback Prevention and Monitoring

Direct Lake mode can unexpectedly fall back to DirectQuery under several conditions, causing significant performance degradation and increased costs. Direct Lake fallback scenarios include exceeding SKU limits, unsupported features, memory pressure, unprocessed tables, and security constraints. Proactive monitoring and prevention strategies ensure consistent Direct Lake performance.

These fallbacks often occur silently, leaving users unaware that their dashboards have switched to slower DirectQuery mode. Implementing comprehensive monitoring and prevention strategies maintains optimal performance while providing visibility into potential issues before they impact user experience.

Alternative approaches include: manual dataset monitoring through Power BI Admin Portal (provides basic visibility but lacks automation), or capacity metrics monitoring for resource utilization (helpful but doesn't address dataset-specific issues).

Optimize Real-time Dashboard Analytics with Microsoft Fabric Mirroring

For BI dashboards requiring real-time analytics on operational databases, Microsoft Fabric Mirroring provides near-zero latency data replication without impacting source system performance. Mirroring creates a read-only analytical copy of your operational database in OneLake, enabling real-time dashboards without complex ETL pipelines.

-- Configure mirroring for real-time dashboard analytics
-- Note: Mirroring is configured through Fabric portal, but queries run against mirrored data

-- Real-time inventory dashboard analytics on mirrored operational database
SELECT 
    p.product_category,
    p.product_name,
    i.current_stock_level,
    i.reorder_point,
    CASE 
        WHEN i.current_stock_level <= i.reorder_point THEN 'Reorder Required'
        WHEN i.current_stock_level <= i.reorder_point * 1.5 THEN 'Low Stock Warning'
        ELSE 'Adequate Stock'
    END as stock_status,
    i.last_updated
FROM mirrored_inventory.products p
INNER JOIN mirrored_inventory.inventory_levels i ON p.product_id = i.product_id
WHERE i.last_updated >= DATEADD(MINUTE, -15, GETDATE())
ORDER BY 
    CASE 
        WHEN i.current_stock_level <= i.reorder_point THEN 1
        WHEN i.current_stock_level <= i.reorder_point * 1.5 THEN 2
        ELSE 3
    END,
    p.product_category;

-- Real-time customer behavior analytics for dashboards
WITH real_time_orders AS (
    SELECT 
        customer_id,
        order_date,
        order_total,
        ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY order_date DESC) as order_rank
    FROM mirrored_orders.orders
    WHERE order_date >= DATEADD(HOUR, -2, GETDATE())  -- Last 2 hours
),

customer_metrics AS (
    SELECT 
        customer_id,
        COUNT(*) as recent_orders,
        SUM(order_total) as recent_revenue,
        AVG(order_total) as avg_order_value,
        MAX(order_date) as last_order_time
    FROM real_time_orders
    GROUP BY customer_id
)

SELECT 
    cm.customer_id,
    cm.recent_orders,
    cm.recent_revenue,
    cm.avg_order_value,
    cm.last_order_time,
    DATEDIFF(MINUTE, cm.last_order_time, GETDATE()) as minutes_since_last_order,
    CASE 
        WHEN cm.recent_orders >= 3 THEN 'High Activity'
        WHEN cm.recent_orders >= 2 THEN 'Moderate Activity'
        ELSE 'Single Purchase'
    END as activity_level
FROM customer_metrics cm
WHERE cm.recent_revenue > 100  -- Focus on high-value recent customers
ORDER BY cm.recent_revenue DESC, cm.last_order_time DESC;

-- Configure mirroring for real-time dashboard analytics
-- Note: Mirroring is configured through Fabric portal, but queries run against mirrored data

-- Real-time inventory dashboard analytics on mirrored operational database
SELECT 
    p.product_category,
    p.product_name,
    i.current_stock_level,
    i.reorder_point,
    CASE 
        WHEN i.current_stock_level <= i.reorder_point THEN 'Reorder Required'
        WHEN i.current_stock_level <= i.reorder_point * 1.5 THEN 'Low Stock Warning'
        ELSE 'Adequate Stock'
    END as stock_status,
    i.last_updated
FROM mirrored_inventory.products p
INNER JOIN mirrored_inventory.inventory_levels i ON p.product_id = i.product_id
WHERE i.last_updated >= DATEADD(MINUTE, -15, GETDATE())
ORDER BY 
    CASE 
        WHEN i.current_stock_level <= i.reorder_point THEN 1
        WHEN i.current_stock_level <= i.reorder_point * 1.5 THEN 2
        ELSE 3
    END,
    p.product_category;

-- Real-time customer behavior analytics for dashboards
WITH real_time_orders AS (
    SELECT 
        customer_id,
        order_date,
        order_total,
        ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY order_date DESC) as order_rank
    FROM mirrored_orders.orders
    WHERE order_date >= DATEADD(HOUR, -2, GETDATE())  -- Last 2 hours
),

customer_metrics AS (
    SELECT 
        customer_id,
        COUNT(*) as recent_orders,
        SUM(order_total) as recent_revenue,
        AVG(order_total) as avg_order_value,
        MAX(order_date) as last_order_time
    FROM real_time_orders
    GROUP BY customer_id
)

SELECT 
    cm.customer_id,
    cm.recent_orders,
    cm.recent_revenue,
    cm.avg_order_value,
    cm.last_order_time,
    DATEDIFF(MINUTE, cm.last_order_time, GETDATE()) as minutes_since_last_order,
    CASE 
        WHEN cm.recent_orders >= 3 THEN 'High Activity'
        WHEN cm.recent_orders >= 2 THEN 'Moderate Activity'
        ELSE 'Single Purchase'
    END as activity_level
FROM customer_metrics cm
WHERE cm.recent_revenue > 100  -- Focus on high-value recent customers
ORDER BY cm.recent_revenue DESC, cm.last_order_time DESC;

-- Configure mirroring for real-time dashboard analytics
-- Note: Mirroring is configured through Fabric portal, but queries run against mirrored data

-- Real-time inventory dashboard analytics on mirrored operational database
SELECT 
    p.product_category,
    p.product_name,
    i.current_stock_level,
    i.reorder_point,
    CASE 
        WHEN i.current_stock_level <= i.reorder_point THEN 'Reorder Required'
        WHEN i.current_stock_level <= i.reorder_point * 1.5 THEN 'Low Stock Warning'
        ELSE 'Adequate Stock'
    END as stock_status,
    i.last_updated
FROM mirrored_inventory.products p
INNER JOIN mirrored_inventory.inventory_levels i ON p.product_id = i.product_id
WHERE i.last_updated >= DATEADD(MINUTE, -15, GETDATE())
ORDER BY 
    CASE 
        WHEN i.current_stock_level <= i.reorder_point THEN 1
        WHEN i.current_stock_level <= i.reorder_point * 1.5 THEN 2
        ELSE 3
    END,
    p.product_category;

-- Real-time customer behavior analytics for dashboards
WITH real_time_orders AS (
    SELECT 
        customer_id,
        order_date,
        order_total,
        ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY order_date DESC) as order_rank
    FROM mirrored_orders.orders
    WHERE order_date >= DATEADD(HOUR, -2, GETDATE())  -- Last 2 hours
),

customer_metrics AS (
    SELECT 
        customer_id,
        COUNT(*) as recent_orders,
        SUM(order_total) as recent_revenue,
        AVG(order_total) as avg_order_value,
        MAX(order_date) as last_order_time
    FROM real_time_orders
    GROUP BY customer_id
)

SELECT 
    cm.customer_id,
    cm.recent_orders,
    cm.recent_revenue,
    cm.avg_order_value,
    cm.last_order_time,
    DATEDIFF(MINUTE, cm.last_order_time, GETDATE()) as minutes_since_last_order,
    CASE 
        WHEN cm.recent_orders >= 3 THEN 'High Activity'
        WHEN cm.recent_orders >= 2 THEN 'Moderate Activity'
        ELSE 'Single Purchase'
    END as activity_level
FROM customer_metrics cm
WHERE cm.recent_revenue > 100  -- Focus on high-value recent customers
ORDER BY cm.recent_revenue DESC, cm.last_order_time DESC;

-- Configure mirroring for real-time dashboard analytics
-- Note: Mirroring is configured through Fabric portal, but queries run against mirrored data

-- Real-time inventory dashboard analytics on mirrored operational database
SELECT 
    p.product_category,
    p.product_name,
    i.current_stock_level,
    i.reorder_point,
    CASE 
        WHEN i.current_stock_level <= i.reorder_point THEN 'Reorder Required'
        WHEN i.current_stock_level <= i.reorder_point * 1.5 THEN 'Low Stock Warning'
        ELSE 'Adequate Stock'
    END as stock_status,
    i.last_updated
FROM mirrored_inventory.products p
INNER JOIN mirrored_inventory.inventory_levels i ON p.product_id = i.product_id
WHERE i.last_updated >= DATEADD(MINUTE, -15, GETDATE())
ORDER BY 
    CASE 
        WHEN i.current_stock_level <= i.reorder_point THEN 1
        WHEN i.current_stock_level <= i.reorder_point * 1.5 THEN 2
        ELSE 3
    END,
    p.product_category;

-- Real-time customer behavior analytics for dashboards
WITH real_time_orders AS (
    SELECT 
        customer_id,
        order_date,
        order_total,
        ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY order_date DESC) as order_rank
    FROM mirrored_orders.orders
    WHERE order_date >= DATEADD(HOUR, -2, GETDATE())  -- Last 2 hours
),

customer_metrics AS (
    SELECT 
        customer_id,
        COUNT(*) as recent_orders,
        SUM(order_total) as recent_revenue,
        AVG(order_total) as avg_order_value,
        MAX(order_date) as last_order_time
    FROM real_time_orders
    GROUP BY customer_id
)

SELECT 
    cm.customer_id,
    cm.recent_orders,
    cm.recent_revenue,
    cm.avg_order_value,
    cm.last_order_time,
    DATEDIFF(MINUTE, cm.last_order_time, GETDATE()) as minutes_since_last_order,
    CASE 
        WHEN cm.recent_orders >= 3 THEN 'High Activity'
        WHEN cm.recent_orders >= 2 THEN 'Moderate Activity'
        ELSE 'Single Purchase'
    END as activity_level
FROM customer_metrics cm
WHERE cm.recent_revenue > 100  -- Focus on high-value recent customers
ORDER BY cm.recent_revenue DESC, cm.last_order_time DESC;

Key benefits of Fabric Mirroring for BI dashboards:

• Zero ETL latency: Data is available for dashboard refresh within seconds of operational changes

• No source system impact: Read-only replica eliminates performance impact on operational databases

• Real-time business intelligence: Enable real-time dashboards without complex streaming architectures

• Simplified architecture: Eliminates need for change data capture (CDC) pipelines and real-time ETL processes

Alternative approaches include: traditional ETL with scheduled refresh (higher latency but more control), DirectQuery to operational databases (real-time but impacts source performance), or composite models with strategic data combinations.

‍

Ad-hoc Analytics Optimization Tactics

Implement Fabric Reflex Auto-Optimization for Dynamic Query Patterns

Ad-hoc analytical queries benefit enormously from Microsoft Fabric's Reflex auto-optimization capabilities, which automatically optimize data layout and statistics based on actual query patterns without manual intervention. Unlike static optimization schemes, Reflex continuously monitors query performance and adapts optimization strategies dynamically.

This proves particularly effective for exploratory data science workflows where analysts pivot between different dimensional combinations unpredictably, as Fabric automatically maintains optimal data organization.

-- Enable auto-optimization for analytical fact table
ALTER TABLE sales_fact_delta SET TBLPROPERTIES (
    'delta.autoOptimize.optimizeWrite' = 'true',
    'delta.autoOptimize.autoCompact' = 'true',
    'delta.tuneFileSizesForRewrites' = 'true',
    'delta.feature.allowColumnDefaults' = 'supported'
);

-- Configure adaptive statistics collection
ALTER TABLE sales_fact_delta SET TBLPROPERTIES (
    'delta.columnMapping.mode' = 'name',
    'delta.enableChangeDataFeed' = 'true',
    'delta.logRetentionDuration' = 'interval 30 days'
);


-- Fabric automatically detects and optimizes common query patterns
INSERT INTO sales_fact_delta
SELECT 
    customer_id,
    product_id,
    c.customer_segment,
    p.product_category,
    sale_date,
    revenue,
    quantity
FROM staging_sales s
JOIN customer_dim c ON s.customer_id = c.customer_id
JOIN product_dim p ON s.product_id = p.product_id;

-- Enable auto-optimization for analytical fact table
ALTER TABLE sales_fact_delta SET TBLPROPERTIES (
    'delta.autoOptimize.optimizeWrite' = 'true',
    'delta.autoOptimize.autoCompact' = 'true',
    'delta.tuneFileSizesForRewrites' = 'true',
    'delta.feature.allowColumnDefaults' = 'supported'
);

-- Configure adaptive statistics collection
ALTER TABLE sales_fact_delta SET TBLPROPERTIES (
    'delta.columnMapping.mode' = 'name',
    'delta.enableChangeDataFeed' = 'true',
    'delta.logRetentionDuration' = 'interval 30 days'
);


-- Fabric automatically detects and optimizes common query patterns
INSERT INTO sales_fact_delta
SELECT 
    customer_id,
    product_id,
    c.customer_segment,
    p.product_category,
    sale_date,
    revenue,
    quantity
FROM staging_sales s
JOIN customer_dim c ON s.customer_id = c.customer_id
JOIN product_dim p ON s.product_id = p.product_id;

-- Enable auto-optimization for analytical fact table
ALTER TABLE sales_fact_delta SET TBLPROPERTIES (
    'delta.autoOptimize.optimizeWrite' = 'true',
    'delta.autoOptimize.autoCompact' = 'true',
    'delta.tuneFileSizesForRewrites' = 'true',
    'delta.feature.allowColumnDefaults' = 'supported'
);

-- Configure adaptive statistics collection
ALTER TABLE sales_fact_delta SET TBLPROPERTIES (
    'delta.columnMapping.mode' = 'name',
    'delta.enableChangeDataFeed' = 'true',
    'delta.logRetentionDuration' = 'interval 30 days'
);


-- Fabric automatically detects and optimizes common query patterns
INSERT INTO sales_fact_delta
SELECT 
    customer_id,
    product_id,
    c.customer_segment,
    p.product_category,
    sale_date,
    revenue,
    quantity
FROM staging_sales s
JOIN customer_dim c ON s.customer_id = c.customer_id
JOIN product_dim p ON s.product_id = p.product_id;

-- Enable auto-optimization for analytical fact table
ALTER TABLE sales_fact_delta SET TBLPROPERTIES (
    'delta.autoOptimize.optimizeWrite' = 'true',
    'delta.autoOptimize.autoCompact' = 'true',
    'delta.tuneFileSizesForRewrites' = 'true',
    'delta.feature.allowColumnDefaults' = 'supported'
);

-- Configure adaptive statistics collection
ALTER TABLE sales_fact_delta SET TBLPROPERTIES (
    'delta.columnMapping.mode' = 'name',
    'delta.enableChangeDataFeed' = 'true',
    'delta.logRetentionDuration' = 'interval 30 days'
);


-- Fabric automatically detects and optimizes common query patterns
INSERT INTO sales_fact_delta
SELECT 
    customer_id,
    product_id,
    c.customer_segment,
    p.product_category,
    sale_date,
    revenue,
    quantity
FROM staging_sales s
JOIN customer_dim c ON s.customer_id = c.customer_id
JOIN product_dim p ON s.product_id = p.product_id;

Alternative approaches include: manual Z-ORDER clustering with periodic maintenance (provides more control but requires operational overhead), traditional table partitioning for predictable access patterns (simpler but less adaptive to changing analytical needs), or Microsoft Fabric Copilot suggestions for automated optimization recommendations.

Optimize Complex Window Function Performance

Analytical workloads frequently require sophisticated window functions for ranking, running totals, lag analysis, and statistical calculations across large datasets. Fabric's Spark SQL optimization for window functions requires careful attention to partitioning strategies and memory management to prevent performance bottlenecks.

The critical factor for window function performance lies in aligning partition keys with analytical access patterns while managing memory allocation for intermediate shuffle operations. Fabric's cost-based optimizer can dramatically improve window function execution when provided with accurate table statistics and proper configuration.

-- Configure session for window function optimization
SET spark.sql.windowExec.buffer.in.memory.threshold = 4096;
SET spark.sql.adaptive.enabled = true;
SET spark.sql.adaptive.coalescePartitions.enabled = true;
SET spark.sql.shuffle.partitions = 400;

-- Complex analytical query with optimized window functions
WITH customer_analytics AS (
    SELECT 
        customer_id,
        sale_date,
        revenue,
        product_category,
        customer_segment,
        -- Running totals and rankings
        SUM(revenue) OVER (
            PARTITION BY customer_id 
            ORDER BY sale_date 
            ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
        ) as customer_lifetime_value,
        
        -- Ranking within segments
        ROW_NUMBER() OVER (
            PARTITION BY customer_segment, product_category 
            ORDER BY revenue DESC
        ) as category_rank,
        
        -- Period-over-period analysis
        LAG(revenue, 1) OVER (
            PARTITION BY customer_id 
            ORDER BY sale_date
        ) as previous_revenue,
        
        -- Statistical functions
        PERCENTILE_CONT(0.5) OVER (
            PARTITION BY customer_segment 
            ORDER BY revenue 
            ROWS BETWEEN 100 PRECEDING AND 100 FOLLOWING
        ) as segment_median_revenue,
        
        -- Advanced analytics
        STDDEV(revenue) OVER (
            PARTITION BY customer_id 
            ORDER BY sale_date 
            ROWS BETWEEN 30 PRECEDING AND CURRENT ROW
        ) as revenue_volatility
        
    FROM sales_fact_delta
    WHERE sale_date >= '2024-01-01'
),

segment_insights AS (
    SELECT 
        customer_segment,
        product_category,
        COUNT(*) as transaction_count,
        AVG(customer_lifetime_value) as avg_clv,
        AVG(revenue_volatility) as avg_volatility,
        -- Complex aggregations with window context
        COUNT(*) OVER (PARTITION BY customer_segment) as segment_size,
        DENSE_RANK() OVER (ORDER BY AVG(customer_lifetime_value) DESC) as clv_rank
    FROM customer_analytics
    GROUP BY customer_segment, product_category
)

SELECT 
    customer_segment,
    product_category,
    transaction_count,
    avg_clv,
    avg_volatility,
    segment_size,
    clv_rank,
    -- Calculate segment contribution
    ROUND(100.0 * transaction_count / segment_size, 2) as category_contribution_pct
FROM segment_insights
WHERE clv_rank <= 10
ORDER BY clv_rank, avg_clv DESC;

-- Configure session for window function optimization
SET spark.sql.windowExec.buffer.in.memory.threshold = 4096;
SET spark.sql.adaptive.enabled = true;
SET spark.sql.adaptive.coalescePartitions.enabled = true;
SET spark.sql.shuffle.partitions = 400;

-- Complex analytical query with optimized window functions
WITH customer_analytics AS (
    SELECT 
        customer_id,
        sale_date,
        revenue,
        product_category,
        customer_segment,
        -- Running totals and rankings
        SUM(revenue) OVER (
            PARTITION BY customer_id 
            ORDER BY sale_date 
            ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
        ) as customer_lifetime_value,
        
        -- Ranking within segments
        ROW_NUMBER() OVER (
            PARTITION BY customer_segment, product_category 
            ORDER BY revenue DESC
        ) as category_rank,
        
        -- Period-over-period analysis
        LAG(revenue, 1) OVER (
            PARTITION BY customer_id 
            ORDER BY sale_date
        ) as previous_revenue,
        
        -- Statistical functions
        PERCENTILE_CONT(0.5) OVER (
            PARTITION BY customer_segment 
            ORDER BY revenue 
            ROWS BETWEEN 100 PRECEDING AND 100 FOLLOWING
        ) as segment_median_revenue,
        
        -- Advanced analytics
        STDDEV(revenue) OVER (
            PARTITION BY customer_id 
            ORDER BY sale_date 
            ROWS BETWEEN 30 PRECEDING AND CURRENT ROW
        ) as revenue_volatility
        
    FROM sales_fact_delta
    WHERE sale_date >= '2024-01-01'
),

segment_insights AS (
    SELECT 
        customer_segment,
        product_category,
        COUNT(*) as transaction_count,
        AVG(customer_lifetime_value) as avg_clv,
        AVG(revenue_volatility) as avg_volatility,
        -- Complex aggregations with window context
        COUNT(*) OVER (PARTITION BY customer_segment) as segment_size,
        DENSE_RANK() OVER (ORDER BY AVG(customer_lifetime_value) DESC) as clv_rank
    FROM customer_analytics
    GROUP BY customer_segment, product_category
)

SELECT 
    customer_segment,
    product_category,
    transaction_count,
    avg_clv,
    avg_volatility,
    segment_size,
    clv_rank,
    -- Calculate segment contribution
    ROUND(100.0 * transaction_count / segment_size, 2) as category_contribution_pct
FROM segment_insights
WHERE clv_rank <= 10
ORDER BY clv_rank, avg_clv DESC;

-- Configure session for window function optimization
SET spark.sql.windowExec.buffer.in.memory.threshold = 4096;
SET spark.sql.adaptive.enabled = true;
SET spark.sql.adaptive.coalescePartitions.enabled = true;
SET spark.sql.shuffle.partitions = 400;

-- Complex analytical query with optimized window functions
WITH customer_analytics AS (
    SELECT 
        customer_id,
        sale_date,
        revenue,
        product_category,
        customer_segment,
        -- Running totals and rankings
        SUM(revenue) OVER (
            PARTITION BY customer_id 
            ORDER BY sale_date 
            ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
        ) as customer_lifetime_value,
        
        -- Ranking within segments
        ROW_NUMBER() OVER (
            PARTITION BY customer_segment, product_category 
            ORDER BY revenue DESC
        ) as category_rank,
        
        -- Period-over-period analysis
        LAG(revenue, 1) OVER (
            PARTITION BY customer_id 
            ORDER BY sale_date
        ) as previous_revenue,
        
        -- Statistical functions
        PERCENTILE_CONT(0.5) OVER (
            PARTITION BY customer_segment 
            ORDER BY revenue 
            ROWS BETWEEN 100 PRECEDING AND 100 FOLLOWING
        ) as segment_median_revenue,
        
        -- Advanced analytics
        STDDEV(revenue) OVER (
            PARTITION BY customer_id 
            ORDER BY sale_date 
            ROWS BETWEEN 30 PRECEDING AND CURRENT ROW
        ) as revenue_volatility
        
    FROM sales_fact_delta
    WHERE sale_date >= '2024-01-01'
),

segment_insights AS (
    SELECT 
        customer_segment,
        product_category,
        COUNT(*) as transaction_count,
        AVG(customer_lifetime_value) as avg_clv,
        AVG(revenue_volatility) as avg_volatility,
        -- Complex aggregations with window context
        COUNT(*) OVER (PARTITION BY customer_segment) as segment_size,
        DENSE_RANK() OVER (ORDER BY AVG(customer_lifetime_value) DESC) as clv_rank
    FROM customer_analytics
    GROUP BY customer_segment, product_category
)

SELECT 
    customer_segment,
    product_category,
    transaction_count,
    avg_clv,
    avg_volatility,
    segment_size,
    clv_rank,
    -- Calculate segment contribution
    ROUND(100.0 * transaction_count / segment_size, 2) as category_contribution_pct
FROM segment_insights
WHERE clv_rank <= 10
ORDER BY clv_rank, avg_clv DESC;

-- Configure session for window function optimization
SET spark.sql.windowExec.buffer.in.memory.threshold = 4096;
SET spark.sql.adaptive.enabled = true;
SET spark.sql.adaptive.coalescePartitions.enabled = true;
SET spark.sql.shuffle.partitions = 400;

-- Complex analytical query with optimized window functions
WITH customer_analytics AS (
    SELECT 
        customer_id,
        sale_date,
        revenue,
        product_category,
        customer_segment,
        -- Running totals and rankings
        SUM(revenue) OVER (
            PARTITION BY customer_id 
            ORDER BY sale_date 
            ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
        ) as customer_lifetime_value,
        
        -- Ranking within segments
        ROW_NUMBER() OVER (
            PARTITION BY customer_segment, product_category 
            ORDER BY revenue DESC
        ) as category_rank,
        
        -- Period-over-period analysis
        LAG(revenue, 1) OVER (
            PARTITION BY customer_id 
            ORDER BY sale_date
        ) as previous_revenue,
        
        -- Statistical functions
        PERCENTILE_CONT(0.5) OVER (
            PARTITION BY customer_segment 
            ORDER BY revenue 
            ROWS BETWEEN 100 PRECEDING AND 100 FOLLOWING
        ) as segment_median_revenue,
        
        -- Advanced analytics
        STDDEV(revenue) OVER (
            PARTITION BY customer_id 
            ORDER BY sale_date 
            ROWS BETWEEN 30 PRECEDING AND CURRENT ROW
        ) as revenue_volatility
        
    FROM sales_fact_delta
    WHERE sale_date >= '2024-01-01'
),

segment_insights AS (
    SELECT 
        customer_segment,
        product_category,
        COUNT(*) as transaction_count,
        AVG(customer_lifetime_value) as avg_clv,
        AVG(revenue_volatility) as avg_volatility,
        -- Complex aggregations with window context
        COUNT(*) OVER (PARTITION BY customer_segment) as segment_size,
        DENSE_RANK() OVER (ORDER BY AVG(customer_lifetime_value) DESC) as clv_rank
    FROM customer_analytics
    GROUP BY customer_segment, product_category
)

SELECT 
    customer_segment,
    product_category,
    transaction_count,
    avg_clv,
    avg_volatility,
    segment_size,
    clv_rank,
    -- Calculate segment contribution
    ROUND(100.0 * transaction_count / segment_size, 2) as category_contribution_pct
FROM segment_insights
WHERE clv_rank <= 10
ORDER BY clv_rank, avg_clv DESC;

Fabric automatically optimizes window function execution by analyzing partition cardinality and choosing between sort-based and hash-based algorithms. When window partitions are small enough to fit in executor memory, Fabric uses in-memory processing to avoid expensive disk spills.

Implement Advanced Delta Lake Time Travel for Analytical Comparison

Analytical workloads frequently require temporal analysis, trend identification, and period-over-period comparisons that benefit from Delta Lake's time travel capabilities. Delta time travel enables sophisticated analytical patterns like point-in-time reconstruction, data quality auditing, and historical trend analysis.

The power of time travel for analytics lies in enabling precise temporal joins and comparisons without maintaining expensive slowly changing dimension tables. Analysts can compare current state against any historical version, enabling sophisticated cohort analysis and trend identification.

What makes this particularly effective is that Delta's transaction log enables efficient time travel queries by maintaining metadata about data file changes over time. Fabric can quickly identify which files contain data for specific versions without scanning entire datasets.

Configure Intelligent Predicate Pushdown and Projection Optimization

Ad-hoc analytical queries often involve complex filtering and column selection patterns that can benefit significantly from advanced predicate pushdown and projection optimization. Fabric's Spark SQL optimizer can dramatically reduce I/O overhead when queries are structured to leverage columnar storage advantages and partition elimination.

The critical insight lies in structuring analytical queries to maximize predicate pushdown effectiveness while minimizing column scan overhead. This becomes particularly important for wide analytical tables with hundreds of columns where analysts typically access only small subsets of data.

When analysts structure queries to leverage these optimizations, query performance improves substantially even on large datasets.

Implement Efficient Data Lifecycle Management with MERGE Operations

Analytical workloads frequently require efficient deletion and update operations for scenarios like GDPR compliance, data corrections, and incremental ETL processes. Microsoft Fabric's Delta Lake MERGE operations provide ACID transaction guarantees while optimizing data lifecycle management through predicate pushdown and file-level optimization.

The critical advantage of MERGE operations lies in enabling efficient data lifecycle management that maintains query performance while handling complex update scenarios including late-arriving updates and out-of-order records.

-- Efficient GDPR compliance with MERGE operations
MERGE INTO customer_analytics_delta AS target
USING (
    SELECT 
        customer_id,
        'DELETED' as customer_status,
        current_timestamp() as deletion_timestamp,
        'GDPR_REQUEST' as deletion_reason
    FROM gdpr_deletion_requests 
    WHERE request_date >= current_date() - interval 30 days
    AND status = 'approved'
) AS source
ON target.customer_id = source.customer_id
WHEN MATCHED THEN
    UPDATE SET
        customer_status = source.customer_status,
        last_modified = source.deletion_timestamp,
        deletion_reason = source.deletion_reason,
        pii_data = NULL,  -- Remove PII fields
        email = NULL,
        phone = NULL
WHEN NOT MATCHED THEN
    INSERT (customer_id, customer_status, last_modified, deletion_reason)
    VALUES (source.customer_id, source.customer_status, source.deletion_timestamp, source.deletion_reason);

-- Optimize MERGE performance with proper clustering
OPTIMIZE customer_analytics_delta ZORDER BY (customer_id, last_modified);

-- Analytical query with efficient filtering of deleted records
SELECT 
    customer_segment,
    region,
    COUNT(*) as active_customers,
    AVG(total_revenue) as avg_customer_value,
    SUM(total_revenue) as segment_revenue
FROM customer_analytics_delta
WHERE customer_status = 'active'  -- Efficiently filters at storage level
AND last_purchase_date >= current_date() - interval 365 days
GROUP BY customer_segment, region
ORDER BY segment_revenue DESC;

-- Efficient GDPR compliance with MERGE operations
MERGE INTO customer_analytics_delta AS target
USING (
    SELECT 
        customer_id,
        'DELETED' as customer_status,
        current_timestamp() as deletion_timestamp,
        'GDPR_REQUEST' as deletion_reason
    FROM gdpr_deletion_requests 
    WHERE request_date >= current_date() - interval 30 days
    AND status = 'approved'
) AS source
ON target.customer_id = source.customer_id
WHEN MATCHED THEN
    UPDATE SET
        customer_status = source.customer_status,
        last_modified = source.deletion_timestamp,
        deletion_reason = source.deletion_reason,
        pii_data = NULL,  -- Remove PII fields
        email = NULL,
        phone = NULL
WHEN NOT MATCHED THEN
    INSERT (customer_id, customer_status, last_modified, deletion_reason)
    VALUES (source.customer_id, source.customer_status, source.deletion_timestamp, source.deletion_reason);

-- Optimize MERGE performance with proper clustering
OPTIMIZE customer_analytics_delta ZORDER BY (customer_id, last_modified);

-- Analytical query with efficient filtering of deleted records
SELECT 
    customer_segment,
    region,
    COUNT(*) as active_customers,
    AVG(total_revenue) as avg_customer_value,
    SUM(total_revenue) as segment_revenue
FROM customer_analytics_delta
WHERE customer_status = 'active'  -- Efficiently filters at storage level
AND last_purchase_date >= current_date() - interval 365 days
GROUP BY customer_segment, region
ORDER BY segment_revenue DESC;

-- Efficient GDPR compliance with MERGE operations
MERGE INTO customer_analytics_delta AS target
USING (
    SELECT 
        customer_id,
        'DELETED' as customer_status,
        current_timestamp() as deletion_timestamp,
        'GDPR_REQUEST' as deletion_reason
    FROM gdpr_deletion_requests 
    WHERE request_date >= current_date() - interval 30 days
    AND status = 'approved'
) AS source
ON target.customer_id = source.customer_id
WHEN MATCHED THEN
    UPDATE SET
        customer_status = source.customer_status,
        last_modified = source.deletion_timestamp,
        deletion_reason = source.deletion_reason,
        pii_data = NULL,  -- Remove PII fields
        email = NULL,
        phone = NULL
WHEN NOT MATCHED THEN
    INSERT (customer_id, customer_status, last_modified, deletion_reason)
    VALUES (source.customer_id, source.customer_status, source.deletion_timestamp, source.deletion_reason);

-- Optimize MERGE performance with proper clustering
OPTIMIZE customer_analytics_delta ZORDER BY (customer_id, last_modified);

-- Analytical query with efficient filtering of deleted records
SELECT 
    customer_segment,
    region,
    COUNT(*) as active_customers,
    AVG(total_revenue) as avg_customer_value,
    SUM(total_revenue) as segment_revenue
FROM customer_analytics_delta
WHERE customer_status = 'active'  -- Efficiently filters at storage level
AND last_purchase_date >= current_date() - interval 365 days
GROUP BY customer_segment, region
ORDER BY segment_revenue DESC;

-- Efficient GDPR compliance with MERGE operations
MERGE INTO customer_analytics_delta AS target
USING (
    SELECT 
        customer_id,
        'DELETED' as customer_status,
        current_timestamp() as deletion_timestamp,
        'GDPR_REQUEST' as deletion_reason
    FROM gdpr_deletion_requests 
    WHERE request_date >= current_date() - interval 30 days
    AND status = 'approved'
) AS source
ON target.customer_id = source.customer_id
WHEN MATCHED THEN
    UPDATE SET
        customer_status = source.customer_status,
        last_modified = source.deletion_timestamp,
        deletion_reason = source.deletion_reason,
        pii_data = NULL,  -- Remove PII fields
        email = NULL,
        phone = NULL
WHEN NOT MATCHED THEN
    INSERT (customer_id, customer_status, last_modified, deletion_reason)
    VALUES (source.customer_id, source.customer_status, source.deletion_timestamp, source.deletion_reason);

-- Optimize MERGE performance with proper clustering
OPTIMIZE customer_analytics_delta ZORDER BY (customer_id, last_modified);

-- Analytical query with efficient filtering of deleted records
SELECT 
    customer_segment,
    region,
    COUNT(*) as active_customers,
    AVG(total_revenue) as avg_customer_value,
    SUM(total_revenue) as segment_revenue
FROM customer_analytics_delta
WHERE customer_status = 'active'  -- Efficiently filters at storage level
AND last_purchase_date >= current_date() - interval 365 days
GROUP BY customer_segment, region
ORDER BY segment_revenue DESC;

Alternative approaches include: MERGE operations for complex update scenarios (provides ACID guarantees but more overhead for simple deletions), partition-based data lifecycle management using date-based retention (simpler but less granular control).

Optimize JOIN strategies for dimensional analysis

Complex analytical queries frequently involve joining large fact tables with multiple dimension tables, where default JOIN strategies can lead to unnecessary data movement and memory pressure. Fabric's Spark SQL JOIN optimization through broadcast hints and bucketing ensures that dimensional analysis queries execute efficiently without shuffle operations.

Once you've set this up, you'll find that analytical queries involving star schema JOINs execute substantially faster because dimension tables are broadcast to all executors, eliminating the shuffle overhead that typically dominates query execution time. Here's where it gets interesting: proper JOIN optimization allows Fabric to perform dimension lookups locally on each executor.

-- Optimize star schema JOINs with broadcast hints
SELECT /*+ BROADCAST(d, p, c) */
    d.date_key,
    d.month_name,
    d.quarter,
    p.product_name,
    p.category,
    c.customer_segment,
    c.region,
    SUM(f.revenue) as total_revenue,
    COUNT(*) as transaction_count,
    AVG(f.revenue) as avg_transaction_value
FROM sales_fact_delta f
INNER JOIN /*+ BROADCAST */ date_dim_delta d ON f.date_key = d.date_key
INNER JOIN /*+ BROADCAST */ product_dim_delta p ON f.product_key = p.product_key
INNER JOIN /*+ BROADCAST */ customer_dim_delta c ON f.customer_key = c.customer_key
WHERE d.fiscal_year = 2024
AND p.category IN ('Electronics', 'Clothing', 'Home')
GROUP BY d.date_key, d.month_name, d.quarter, p.product_name, p.category, 
         c.customer_segment, c.region
HAVING SUM(f.revenue) > 10000
ORDER BY total_revenue DESC;

-- Alternative: Use bucketed tables for large dimension scenarios
CREATE TABLE sales_fact_bucketed
USING DELTA
CLUSTERED BY (customer_key) INTO 32 BUCKETS
AS SELECT * FROM sales_fact_delta;

CREATE TABLE customer_dim_bucketed  
USING DELTA
CLUSTERED BY (customer_key) INTO 32 BUCKETS
AS SELECT * FROM customer_dim_delta;

-- Optimize star schema JOINs with broadcast hints
SELECT /*+ BROADCAST(d, p, c) */
    d.date_key,
    d.month_name,
    d.quarter,
    p.product_name,
    p.category,
    c.customer_segment,
    c.region,
    SUM(f.revenue) as total_revenue,
    COUNT(*) as transaction_count,
    AVG(f.revenue) as avg_transaction_value
FROM sales_fact_delta f
INNER JOIN /*+ BROADCAST */ date_dim_delta d ON f.date_key = d.date_key
INNER JOIN /*+ BROADCAST */ product_dim_delta p ON f.product_key = p.product_key
INNER JOIN /*+ BROADCAST */ customer_dim_delta c ON f.customer_key = c.customer_key
WHERE d.fiscal_year = 2024
AND p.category IN ('Electronics', 'Clothing', 'Home')
GROUP BY d.date_key, d.month_name, d.quarter, p.product_name, p.category, 
         c.customer_segment, c.region
HAVING SUM(f.revenue) > 10000
ORDER BY total_revenue DESC;

-- Alternative: Use bucketed tables for large dimension scenarios
CREATE TABLE sales_fact_bucketed
USING DELTA
CLUSTERED BY (customer_key) INTO 32 BUCKETS
AS SELECT * FROM sales_fact_delta;

CREATE TABLE customer_dim_bucketed  
USING DELTA
CLUSTERED BY (customer_key) INTO 32 BUCKETS
AS SELECT * FROM customer_dim_delta;

-- Optimize star schema JOINs with broadcast hints
SELECT /*+ BROADCAST(d, p, c) */
    d.date_key,
    d.month_name,
    d.quarter,
    p.product_name,
    p.category,
    c.customer_segment,
    c.region,
    SUM(f.revenue) as total_revenue,
    COUNT(*) as transaction_count,
    AVG(f.revenue) as avg_transaction_value
FROM sales_fact_delta f
INNER JOIN /*+ BROADCAST */ date_dim_delta d ON f.date_key = d.date_key
INNER JOIN /*+ BROADCAST */ product_dim_delta p ON f.product_key = p.product_key
INNER JOIN /*+ BROADCAST */ customer_dim_delta c ON f.customer_key = c.customer_key
WHERE d.fiscal_year = 2024
AND p.category IN ('Electronics', 'Clothing', 'Home')
GROUP BY d.date_key, d.month_name, d.quarter, p.product_name, p.category, 
         c.customer_segment, c.region
HAVING SUM(f.revenue) > 10000
ORDER BY total_revenue DESC;

-- Alternative: Use bucketed tables for large dimension scenarios
CREATE TABLE sales_fact_bucketed
USING DELTA
CLUSTERED BY (customer_key) INTO 32 BUCKETS
AS SELECT * FROM sales_fact_delta;

CREATE TABLE customer_dim_bucketed  
USING DELTA
CLUSTERED BY (customer_key) INTO 32 BUCKETS
AS SELECT * FROM customer_dim_delta;

-- Optimize star schema JOINs with broadcast hints
SELECT /*+ BROADCAST(d, p, c) */
    d.date_key,
    d.month_name,
    d.quarter,
    p.product_name,
    p.category,
    c.customer_segment,
    c.region,
    SUM(f.revenue) as total_revenue,
    COUNT(*) as transaction_count,
    AVG(f.revenue) as avg_transaction_value
FROM sales_fact_delta f
INNER JOIN /*+ BROADCAST */ date_dim_delta d ON f.date_key = d.date_key
INNER JOIN /*+ BROADCAST */ product_dim_delta p ON f.product_key = p.product_key
INNER JOIN /*+ BROADCAST */ customer_dim_delta c ON f.customer_key = c.customer_key
WHERE d.fiscal_year = 2024
AND p.category IN ('Electronics', 'Clothing', 'Home')
GROUP BY d.date_key, d.month_name, d.quarter, p.product_name, p.category, 
         c.customer_segment, c.region
HAVING SUM(f.revenue) > 10000
ORDER BY total_revenue DESC;

-- Alternative: Use bucketed tables for large dimension scenarios
CREATE TABLE sales_fact_bucketed
USING DELTA
CLUSTERED BY (customer_key) INTO 32 BUCKETS
AS SELECT * FROM sales_fact_delta;

CREATE TABLE customer_dim_bucketed  
USING DELTA
CLUSTERED BY (customer_key) INTO 32 BUCKETS
AS SELECT * FROM customer_dim_delta;

Alternative approaches include: sort-merge JOIN optimization for large table combinations (handles bigger datasets but requires sorted data), or denormalized fact table designs to avoid JOINs entirely (faster queries but increased storage and update complexity).

Leverage KQL Database for Advanced Time-Series Analytics

Microsoft Fabric KQL Database is specifically optimized for time-series and telemetry data analytics, providing superior performance for complex analytical queries, log analytics, IoT data, and operational monitoring use cases compared to traditional SQL engines.

Key benefits of KQL Database for advanced analytics:

• Time-series optimization: Native support for time-series data patterns and functions

• High-performance aggregations: Optimized for complex analytical queries across large datasets

• Materialized views: Automatic query acceleration through pre-computed aggregations

• Advanced analytics: Built-in statistical functions and anomaly detection capabilities

• Hybrid integration: Seamless combination with Lakehouse data for comprehensive analysis

Alternative approaches include: Spark SQL with Delta tables for time-series (more general but less optimized), traditional data warehouses with time-series extensions (familiar but less performant), or specialized time-series databases (optimal but requires separate infrastructure).

‍

ETL/Streaming Optimization Tactics

Implement Structured Streaming with Watermarks

Real-time ETL pipelines often struggle with late-arriving data and memory accumulation in stateful operations like window aggregations and stream-to-stream JOINs. Structured Streaming watermarks enable efficient state management by defining how long to wait for late data before finalizing aggregation results.

# Implement watermarked streaming aggregations for ETL pipeline
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

# Define schema for streaming data
sales_schema = StructType([
    StructField("transaction_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("revenue", DecimalType(10,2), True),
    StructField("event_timestamp", TimestampType(), True)
])

# Read streaming data with watermark configuration
streaming_sales = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "your-kafka-cluster") \
    .option("subscribe", "sales-topic") \
    .load() \
    .select(from_json(col("value").cast("string"), sales_schema).alias("data")) \
    .select("data.*") \
    .withWatermark("event_timestamp", "10 minutes")

# Perform windowed aggregations with late data handling
windowed_sales = streaming_sales \
    .groupBy(
        window(col("event_timestamp"), "5 minutes", "1 minute"),
        col("product_id")
    ) \
    .agg(
        sum("revenue").alias("total_revenue"),
        count("*").alias("transaction_count"),
        avg("revenue").alias("avg_revenue")
    ) \
    .select(
        col("window.start").alias("window_start"),
        col("window.end").alias("window_end"),
        col("product_id"),
        col("total_revenue"),
        col("transaction_count"),
        col("avg_revenue")
    )

# Write results to Delta table with watermark-based state cleanup
query = windowed_sales.writeStream \
    .format("delta") \
    .option("checkpointLocation", "/lakehouse/checkpoints/windowed_sales") \
    .outputMode("append") \
    .trigger(processingTime="30 seconds") \
    .toTable("real_time_sales_agg")

# Implement watermarked streaming aggregations for ETL pipeline
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

# Define schema for streaming data
sales_schema = StructType([
    StructField("transaction_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("revenue", DecimalType(10,2), True),
    StructField("event_timestamp", TimestampType(), True)
])

# Read streaming data with watermark configuration
streaming_sales = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "your-kafka-cluster") \
    .option("subscribe", "sales-topic") \
    .load() \
    .select(from_json(col("value").cast("string"), sales_schema).alias("data")) \
    .select("data.*") \
    .withWatermark("event_timestamp", "10 minutes")

# Perform windowed aggregations with late data handling
windowed_sales = streaming_sales \
    .groupBy(
        window(col("event_timestamp"), "5 minutes", "1 minute"),
        col("product_id")
    ) \
    .agg(
        sum("revenue").alias("total_revenue"),
        count("*").alias("transaction_count"),
        avg("revenue").alias("avg_revenue")
    ) \
    .select(
        col("window.start").alias("window_start"),
        col("window.end").alias("window_end"),
        col("product_id"),
        col("total_revenue"),
        col("transaction_count"),
        col("avg_revenue")
    )

# Write results to Delta table with watermark-based state cleanup
query = windowed_sales.writeStream \
    .format("delta") \
    .option("checkpointLocation", "/lakehouse/checkpoints/windowed_sales") \
    .outputMode("append") \
    .trigger(processingTime="30 seconds") \
    .toTable("real_time_sales_agg")

# Implement watermarked streaming aggregations for ETL pipeline
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

# Define schema for streaming data
sales_schema = StructType([
    StructField("transaction_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("revenue", DecimalType(10,2), True),
    StructField("event_timestamp", TimestampType(), True)
])

# Read streaming data with watermark configuration
streaming_sales = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "your-kafka-cluster") \
    .option("subscribe", "sales-topic") \
    .load() \
    .select(from_json(col("value").cast("string"), sales_schema).alias("data")) \
    .select("data.*") \
    .withWatermark("event_timestamp", "10 minutes")

# Perform windowed aggregations with late data handling
windowed_sales = streaming_sales \
    .groupBy(
        window(col("event_timestamp"), "5 minutes", "1 minute"),
        col("product_id")
    ) \
    .agg(
        sum("revenue").alias("total_revenue"),
        count("*").alias("transaction_count"),
        avg("revenue").alias("avg_revenue")
    ) \
    .select(
        col("window.start").alias("window_start"),
        col("window.end").alias("window_end"),
        col("product_id"),
        col("total_revenue"),
        col("transaction_count"),
        col("avg_revenue")
    )

# Write results to Delta table with watermark-based state cleanup
query = windowed_sales.writeStream \
    .format("delta") \
    .option("checkpointLocation", "/lakehouse/checkpoints/windowed_sales") \
    .outputMode("append") \
    .trigger(processingTime="30 seconds") \
    .toTable("real_time_sales_agg")

# Implement watermarked streaming aggregations for ETL pipeline
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

# Define schema for streaming data
sales_schema = StructType([
    StructField("transaction_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("revenue", DecimalType(10,2), True),
    StructField("event_timestamp", TimestampType(), True)
])

# Read streaming data with watermark configuration
streaming_sales = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "your-kafka-cluster") \
    .option("subscribe", "sales-topic") \
    .load() \
    .select(from_json(col("value").cast("string"), sales_schema).alias("data")) \
    .select("data.*") \
    .withWatermark("event_timestamp", "10 minutes")

# Perform windowed aggregations with late data handling
windowed_sales = streaming_sales \
    .groupBy(
        window(col("event_timestamp"), "5 minutes", "1 minute"),
        col("product_id")
    ) \
    .agg(
        sum("revenue").alias("total_revenue"),
        count("*").alias("transaction_count"),
        avg("revenue").alias("avg_revenue")
    ) \
    .select(
        col("window.start").alias("window_start"),
        col("window.end").alias("window_end"),
        col("product_id"),
        col("total_revenue"),
        col("transaction_count"),
        col("avg_revenue")
    )

# Write results to Delta table with watermark-based state cleanup
query = windowed_sales.writeStream \
    .format("delta") \
    .option("checkpointLocation", "/lakehouse/checkpoints/windowed_sales") \
    .outputMode("append") \
    .trigger(processingTime="30 seconds") \
    .toTable("real_time_sales_agg")