Real-Time Event Platform: Azure Event Hubs + Stream Analytics + Cosmos DB + SignalR

Introduction: The Real-Time Imperative

Batch processing is no longer sufficient for modern businesses. Fraud detection, IoT monitoring, live dashboards, and operational intelligence all demand sub-second event processing. This deep dive builds a complete real-time event platform capable of ingesting 10 million events per second, processing them through windowed aggregations and pattern detection, storing results in Cosmos DB for sub-10ms reads, and pushing live updates to browser dashboards via SignalR.

Prerequisites

• Azure subscription with Event Hubs Premium or Dedicated access
• Azure Stream Analytics cluster
• Azure Cosmos DB account (NoSQL API)
• Azure SignalR Service
• .NET 8 SDK for producers and consumers
• Azure Functions for serverless processing

Phase 1: Event Hubs Ingestion Layer

Provisioning High-Throughput Event Hubs

param location string = resourceGroup().location

resource eventHubNamespace 'Microsoft.EventHub/namespaces@2024-01-01' = {
  name: 'eh-realtime-prod'
  location: location
  sku: {
    name: 'Premium'
    tier: 'Premium'
    capacity: 4
  }
  properties: {
    minimumTlsVersion: '1.2'
    publicNetworkAccess: 'SecuredByPerimeter'
    zoneRedundant: true
    kafkaEnabled: true
    disableLocalAuth: true
  }
}

resource telemetryHub 'Microsoft.EventHub/namespaces/eventhubs@2024-01-01' = {
  parent: eventHubNamespace
  name: 'device-telemetry'
  properties: {
    partitionCount: 32
    messageRetentionInDays: 7
    status: 'Active'
    captureDescription: {
      enabled: true
      encoding: 'Avro'
      intervalInSeconds: 300
      sizeLimitInBytes: 314572800
      destination: {
        name: 'EventHubArchive.AzureBlockBlob'
        properties: {
          storageAccountResourceId: storageAccount.id
          blobContainer: 'event-archive'
          archiveNameFormat: '{Namespace}/{EventHub}/p={PartitionId}/y={Year}/m={Month}/d={Day}/h={Hour}/{Minute}/{Second}'
        }
      }
    }
  }
}

resource alertsHub 'Microsoft.EventHub/namespaces/eventhubs@2024-01-01' = {
  parent: eventHubNamespace
  name: 'processed-alerts'
  properties: {
    partitionCount: 8
    messageRetentionInDays: 3
  }
}

// Schema Registry for event validation
resource schemaGroup 'Microsoft.EventHub/namespaces/schemagroups@2024-01-01' = {
  parent: eventHubNamespace
  name: 'telemetry-schemas'
  properties: {
    schemaType: 'Avro'
    schemaCompatibility: 'Backward'
    groupProperties: {}
  }
}

High-Performance Event Producer

using Azure.Messaging.EventHubs;
using Azure.Messaging.EventHubs.Producer;
using System.Text.Json;

public class TelemetryProducer : IAsyncDisposable
{
    private readonly EventHubBufferedProducerClient _producer;
    private readonly ILogger<TelemetryProducer> _logger;
    private long _totalSent;
    private long _totalFailed;

    public TelemetryProducer(string connectionString, ILogger<TelemetryProducer> logger)
    {
        _logger = logger;
        _producer = new EventHubBufferedProducerClient(connectionString, "device-telemetry",
            new EventHubBufferedProducerClientOptions
            {
                MaximumWaitTime = TimeSpan.FromMilliseconds(500),
                MaximumEventBufferLengthPerPartition = 5000,
                MaximumConcurrentSendsPerPartition = 2
            });

        _producer.SendEventBatchSucceededAsync += args =>
        {
            Interlocked.Add(ref _totalSent, args.EventBatch.Count);
            return Task.CompletedTask;
        };

        _producer.SendEventBatchFailedAsync += args =>
        {
            Interlocked.Add(ref _totalFailed, args.EventBatch.Count);
            _logger.LogError(args.Exception, "Batch send failed for partition {Partition}", args.PartitionId);
            return Task.CompletedTask;
        };
    }

    public async Task SendTelemetryAsync(DeviceTelemetry telemetry)
    {
        var eventData = new EventData(JsonSerializer.SerializeToUtf8Bytes(telemetry));
        eventData.Properties["DeviceType"] = telemetry.DeviceType;
        eventData.Properties["Region"] = telemetry.Region;
        eventData.ContentType = "application/json";

        await _producer.EnqueueEventAsync(eventData, new EnqueueEventOptions
        {
            PartitionKey = telemetry.DeviceId
        });
    }

    public async ValueTask DisposeAsync()
    {
        await _producer.FlushAsync();
        await _producer.DisposeAsync();
        _logger.LogInformation("Producer closed. Sent: {Sent}, Failed: {Failed}", _totalSent, _totalFailed);
    }
}

public record DeviceTelemetry(
    string DeviceId,
    string DeviceType,
    string Region,
    DateTime Timestamp,
    double Temperature,
    double Humidity,
    double Pressure,
    Dictionary<string, double> CustomMetrics
);

Phase 2: Stream Analytics Complex Event Processing

Real-Time Aggregation and Anomaly Detection

-- Stream Analytics Job: device-telemetry-processor

-- Input: Event Hub (device-telemetry)
-- Output 1: Cosmos DB (aggregated metrics)
-- Output 2: Event Hub (processed-alerts)
-- Output 3: Power BI (live dashboard dataset)

-- Tumbling window: 1-minute device averages
SELECT
    DeviceId,
    DeviceType,
    Region,
    System.Timestamp() AS WindowEnd,
    COUNT(*) AS EventCount,
    AVG(Temperature) AS AvgTemperature,
    MIN(Temperature) AS MinTemperature,
    MAX(Temperature) AS MaxTemperature,
    STDEV(Temperature) AS StdDevTemperature,
    AVG(Humidity) AS AvgHumidity,
    AVG(Pressure) AS AvgPressure,
    PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY Temperature) 
        OVER (PARTITION BY DeviceId) AS P95Temperature
INTO [cosmos-aggregates]
FROM [device-telemetry] TIMESTAMP BY Timestamp
GROUP BY
    DeviceId, DeviceType, Region,
    TumblingWindow(minute, 1)

-- Sliding window: Anomaly detection (temperature spike)
SELECT
    DeviceId,
    DeviceType,
    Region,
    System.Timestamp() AS AlertTime,
    'TemperatureSpike' AS AlertType,
    AVG(Temperature) AS CurrentAvg,
    LAG(AVG(Temperature), 5) OVER (
        PARTITION BY DeviceId 
        LIMIT DURATION(minute, 10)
    ) AS PreviousAvg,
    CASE
        WHEN AVG(Temperature) > LAG(AVG(Temperature), 5) OVER (
            PARTITION BY DeviceId LIMIT DURATION(minute, 10)
        ) * 1.5 THEN 'Critical'
        WHEN AVG(Temperature) > LAG(AVG(Temperature), 5) OVER (
            PARTITION BY DeviceId LIMIT DURATION(minute, 10)
        ) * 1.2 THEN 'Warning'
        ELSE 'Info'
    END AS Severity
INTO [alerts-output]
FROM [device-telemetry] TIMESTAMP BY Timestamp
GROUP BY
    DeviceId, DeviceType, Region,
    SlidingWindow(minute, 5)
HAVING
    AVG(Temperature) > LAG(AVG(Temperature), 5) OVER (
        PARTITION BY DeviceId LIMIT DURATION(minute, 10)
    ) * 1.2

-- Session window: Device connectivity tracking
SELECT
    DeviceId,
    DeviceType,
    MIN(Timestamp) AS SessionStart,
    MAX(Timestamp) AS SessionEnd,
    DATEDIFF(second, MIN(Timestamp), MAX(Timestamp)) AS SessionDurationSec,
    COUNT(*) AS EventsInSession
INTO [cosmos-sessions]
FROM [device-telemetry] TIMESTAMP BY Timestamp
GROUP BY
    DeviceId, DeviceType,
    SessionWindow(minute, 2, 60)

-- Geofence detection: Devices leaving allowed regions
WITH DeviceLocations AS (
    SELECT
        DeviceId,
        Latitude,
        Longitude,
        System.Timestamp() AS EventTime
    FROM [device-telemetry] TIMESTAMP BY Timestamp
    WHERE Latitude IS NOT NULL AND Longitude IS NOT NULL
)
SELECT
    d.DeviceId,
    d.Latitude,
    d.Longitude,
    d.EventTime,
    'GeofenceViolation' AS AlertType,
    g.FenceName,
    ST_DISTANCE(
        CreatePoint(d.Latitude, d.Longitude),
        CreatePoint(g.CenterLat, g.CenterLon)
    ) AS DistanceFromCenter
INTO [alerts-output]
FROM DeviceLocations d
CROSS JOIN [geofence-reference] g
WHERE ST_DISTANCE(
    CreatePoint(d.Latitude, d.Longitude),
    CreatePoint(g.CenterLat, g.CenterLon)
) > g.RadiusMeters

Phase 3: Cosmos DB for Low-Latency Storage

Container Configuration for Time-Series Data

using Microsoft.Azure.Cosmos;

public class CosmosTimeSeriesRepository
{
    private readonly Container _aggregatesContainer;
    private readonly Container _alertsContainer;

    public CosmosTimeSeriesRepository(CosmosClient client)
    {
        var database = client.GetDatabase("realtime-analytics");
        _aggregatesContainer = database.GetContainer("device-aggregates");
        _alertsContainer = database.GetContainer("alerts");
    }

    public async Task UpsertAggregateAsync(DeviceAggregate aggregate)
    {
        // Partition key: DeviceId for hot partition monitoring
        // TTL: 30 days for 1-minute aggregates, 365 days for hourly rollups
        aggregate.Ttl = aggregate.GranularityMinutes switch
        {
            1 => 30 * 24 * 3600,    // 30 days
            60 => 365 * 24 * 3600,   // 1 year
            _ => -1                   // No expiry
        };

        await _aggregatesContainer.UpsertItemAsync(
            aggregate,
            new PartitionKey(aggregate.DeviceId),
            new ItemRequestOptions
            {
                EnableContentResponseOnWrite = false
            });
    }

    public async Task<IReadOnlyList<DeviceAggregate>> GetRecentAggregatesAsync(
        string deviceId, int minutes = 60)
    {
        var cutoff = DateTime.UtcNow.AddMinutes(-minutes);
        var query = new QueryDefinition(
            "SELECT * FROM c WHERE c.deviceId = @deviceId AND c.windowEnd >= @cutoff ORDER BY c.windowEnd DESC")
            .WithParameter("@deviceId", deviceId)
            .WithParameter("@cutoff", cutoff);

        var results = new List<DeviceAggregate>();
        using var iterator = _aggregatesContainer.GetItemQueryIterator<DeviceAggregate>(
            query, requestOptions: new QueryRequestOptions
            {
                PartitionKey = new PartitionKey(deviceId),
                MaxItemCount = 100
            });

        while (iterator.HasMoreResults)
        {
            var response = await iterator.ReadNextAsync();
            results.AddRange(response);
        }

        return results;
    }
}

Change Feed for Real-Time Notifications

public class AlertChangeFeedProcessor
{
    public async Task StartAsync(CosmosClient client, IHubContext<DashboardHub> hubContext)
    {
        var leaseContainer = client.GetContainer("realtime-analytics", "leases");
        var alertsContainer = client.GetContainer("realtime-analytics", "alerts");

        var processor = alertsContainer
            .GetChangeFeedProcessorBuilder<AlertDocument>("alert-processor",
                async (IReadOnlyCollection<AlertDocument> changes, CancellationToken ct) =>
                {
                    foreach (var alert in changes)
                    {
                        // Push to SignalR clients in real-time
                        await hubContext.Clients
                            .Group($"region-{alert.Region}")
                            .SendAsync("NewAlert", new
                            {
                                alert.DeviceId,
                                alert.AlertType,
                                alert.Severity,
                                alert.AlertTime,
                                alert.CurrentAvg,
                                alert.PreviousAvg
                            }, ct);

                        // Push to device-specific subscribers
                        await hubContext.Clients
                            .Group($"device-{alert.DeviceId}")
                            .SendAsync("DeviceAlert", alert, ct);
                    }
                })
            .WithInstanceName(Environment.MachineName)
            .WithLeaseContainer(leaseContainer)
            .WithStartTime(DateTime.UtcNow.AddMinutes(-5))
            .WithMaxItems(50)
            .WithPollInterval(TimeSpan.FromMilliseconds(500))
            .Build();

        await processor.StartAsync();
    }
}

Phase 4: SignalR Live Dashboard

SignalR Hub for Real-Time Updates

using Microsoft.AspNetCore.SignalR;

public class DashboardHub : Hub
{
    private readonly CosmosTimeSeriesRepository _repository;

    public DashboardHub(CosmosTimeSeriesRepository repository)
    {
        _repository = repository;
    }

    public async Task SubscribeToRegion(string region)
    {
        await Groups.AddToGroupAsync(Context.ConnectionId, $"region-{region}");

        // Send initial state
        var recentAlerts = await _repository.GetRecentAlertsByRegionAsync(region, 30);
        await Clients.Caller.SendAsync("InitialAlerts", recentAlerts);
    }

    public async Task SubscribeToDevice(string deviceId)
    {
        await Groups.AddToGroupAsync(Context.ConnectionId, $"device-{deviceId}");

        // Send recent telemetry
        var recentData = await _repository.GetRecentAggregatesAsync(deviceId, 60);
        await Clients.Caller.SendAsync("InitialTelemetry", recentData);
    }

    public async Task UnsubscribeFromRegion(string region)
    {
        await Groups.RemoveFromGroupAsync(Context.ConnectionId, $"region-{region}");
    }

    public override async Task OnDisconnectedAsync(Exception? exception)
    {
        // Clean up happens automatically via SignalR group management
        await base.OnDisconnectedAsync(exception);
    }
}

Client-Side Dashboard Integration

// dashboard.js - Real-time dashboard client
const connection = new signalR.HubConnectionBuilder()
    .withUrl("/hubs/dashboard")
    .withAutomaticReconnect([0, 2000, 5000, 10000, 30000])
    .configureLogging(signalR.LogLevel.Warning)
    .build();

// Handle real-time telemetry updates
connection.on("TelemetryUpdate", (data) => {
    updateChart(data.deviceId, {
        timestamp: data.windowEnd,
        temperature: data.avgTemperature,
        humidity: data.avgHumidity,
        pressure: data.avgPressure,
        eventCount: data.eventCount
    });
    updateKPI("totalEvents", data.eventCount);
});

// Handle alerts
connection.on("NewAlert", (alert) => {
    const alertElement = createAlertCard(alert);
    document.getElementById("alert-feed").prepend(alertElement);

    if (alert.severity === "Critical") {
        playAlertSound();
        showNotification(`Critical: ${alert.alertType} on ${alert.deviceId}`);
    }

    updateKPI("activeAlerts", 1, "increment");
});

// Connection lifecycle
connection.onreconnecting(() => {
    document.getElementById("status").textContent = "Reconnecting...";
    document.getElementById("status").className = "status-warning";
});

connection.onreconnected(() => {
    document.getElementById("status").textContent = "Connected";
    document.getElementById("status").className = "status-ok";
    // Re-subscribe to groups after reconnection
    subscribeToActiveRegions();
});

async function start() {
    try {
        await connection.start();
        await connection.invoke("SubscribeToRegion", "us-east");
        await connection.invoke("SubscribeToRegion", "eu-west");
    } catch (err) {
        console.error("SignalR connection error:", err);
        setTimeout(start, 5000);
    }
}

start();

Performance Benchmarks

Best Practices

1. Partition keys matter: Use device ID for Event Hubs partitioning to preserve ordering per device
2. Use Event Hubs Capture for cold path: Archive all events to ADLS for reprocessing and compliance
3. Schema Registry for evolution: Validate event schemas before processing to prevent pipeline failures
4. Cosmos DB TTL for cost control: Auto-expire old aggregates to keep storage costs predictable
5. SignalR groups over broadcast: Only send data to clients that need it to reduce bandwidth
6. Watermarks in Stream Analytics: Handle late-arriving events with proper watermark policies

Architecture Decision and Tradeoffs

When designing integrated solutions solutions with Azure + Power Platform, consider these key architectural trade-offs:

Recommendation: Start with the managed approach for most workloads and move to custom only when specific requirements demand it.

Validation and Versioning

• Last validated: April 2026
• Validate examples against your tenant, region, and SKU constraints before production rollout.
• Keep module, CLI, and SDK versions pinned in automation pipelines and review quarterly.

Security and Governance Considerations

• Apply least-privilege access using RBAC roles and just-in-time elevation for admin tasks.
• Store secrets in managed secret stores and avoid embedding credentials in scripts or source files.
• Enable audit logging, data protection policies, and periodic access reviews for regulated workloads.

Cost and Performance Notes

• Define budgets and alerts, then monitor usage and cost trends continuously after go-live.
• Baseline performance with synthetic and real-user checks before and after major changes.
• Scale resources with measured thresholds and revisit sizing after usage pattern changes.

Official Microsoft References

• https://learn.microsoft.com/azure/architecture/
• https://learn.microsoft.com/azure/well-architected/
• https://learn.microsoft.com/power-platform/guidance/

Public Examples from Official Sources

• These examples are sourced from official public Microsoft documentation and sample repositories.
• Documentation examples: https://learn.microsoft.com/azure/well-architected/
• Sample repositories: https://github.com/Azure/ArchitectureCenter
• Prefer adapting these examples to your tenant, subscriptions, and governance requirements before production use.

Key Takeaways

• Event Hubs Premium provides the throughput and reliability needed for enterprise event processing
• Stream Analytics windowed queries enable real-time aggregation, anomaly detection, and pattern matching
• Cosmos DB change feed creates a reactive pipeline from storage to live dashboards
• SignalR delivers sub-second updates to thousands of concurrent dashboard users
• Proper partitioning and TTL strategies keep the platform performant and cost-effective at scale

Further Reading

• Azure Event Hubs Documentation
• Stream Analytics Query Language
• Cosmos DB Change Feed
• Azure SignalR Service