Azure Event Hubs
Complete Guide

// Events with same partition key → same partition → ordered
var eventData = new EventData(Encoding.UTF8.GetBytes(JsonSerializer.Serialize(telemetry)))
{
    Properties =
    {
        ["DeviceId"] = "sensor-42",
        ["Region"]   = "EU-West"
    }
};

// Send with partition key — all events from same device go to same partition
await producerClient.SendAsync(
    new[] { eventData },
    new SendEventOptions { PartitionKey = "sensor-42" });

var connectionString = "Endpoint=sb://<namespace>.servicebus.windows.net/;...";
var eventHubName = "telemetry";

await using var producerClient = new EventHubProducerClient(connectionString, eventHubName);

// Create a batch (respects size limits automatically)
using EventDataBatch eventBatch = await producerClient.CreateBatchAsync();

for (int i = 0; i < 100; i++)
{
    var payload = JsonSerializer.Serialize(new { deviceId = "d1", temp = 22.5 + i, ts = DateTime.UtcNow });
    var eventData = new EventData(Encoding.UTF8.GetBytes(payload));
    eventData.Properties["EventType"] = "TemperatureReading";

    if (!eventBatch.TryAdd(eventData))
    {
        // Batch full — send current and start new
        await producerClient.SendAsync(eventBatch);
    }
}

await producerClient.SendAsync(eventBatch);

var storageClient = new BlobContainerClient(storageConnStr, "checkpoints");
var processorClient = new EventProcessorClient(
    storageClient,
    "$Default",          // consumer group
    connectionString,
    eventHubName);

processorClient.ProcessEventAsync += async args =>
{
    string body = args.Data.EventBody.ToString();
    Console.WriteLine($"Partition {args.Partition.PartitionId}: {body}");
    Console.WriteLine($"Offset: {args.Data.Offset}, SeqNo: {args.Data.SequenceNumber}");

    // Checkpoint every N events or on a timer
    await args.UpdateCheckpointAsync();
};

processorClient.ProcessErrorAsync += args =>
{
    Console.WriteLine($"Error on partition {args.PartitionId}: {args.Exception}");
    return Task.CompletedTask;
};

await processorClient.StartProcessingAsync();
Console.ReadKey();
await processorClient.StopProcessingAsync();

from azure.eventhub import EventHubProducerClient, EventData
import json

producer = EventHubProducerClient.from_connection_string(
    conn_str="Endpoint=sb://...",
    eventhub_name="telemetry"
)

with producer:
    event_data_batch = producer.create_batch(partition_key="sensor-42")
    for i in range(100):
        payload = json.dumps({"deviceId": "sensor-42", "temp": 22.5 + i})
        event_data_batch.add(EventData(payload))
    producer.send_batch(event_data_batch)

az eventhubs eventhub consumer-group create \
  --resource-group myRG \
  --namespace-name myNamespace \
  --eventhub-name telemetry \
  --name analytics-cg

resource eventHub 'Microsoft.EventHub/namespaces/eventhubs@2022-01-01-preview' = {
  name: '${namespaceName}/${eventHubName}'
  properties: {
    partitionCount: 4
    messageRetentionInDays: 7
    captureDescription: {
      enabled: true
      encoding: 'Avro'
      intervalInSeconds: 300
      sizeLimitInBytes: 314572800
      skipEmptyArchives: true
      destination: {
        name: 'EventHubArchive.AzureBlockBlob'
        properties: {
          storageAccountResourceId: storageAccount.id
          blobContainer: 'eventhub-archive'
          archiveNameFormat: '{Namespace}/{EventHub}/{PartitionId}/{Year}/{Month}/{Day}/{Hour}/{Minute}/{Second}'
        }
      }
    }
  }
}

var schemaRegistryClient = new SchemaRegistryClient(
    "<namespace>.servicebus.windows.net",
    new DefaultAzureCredential());

var serializer = new SchemaRegistryAvroSerializer(
    schemaRegistryClient,
    schemaGroupName: "my-schema-group",
    new SchemaRegistryAvroSerializerOptions { AutoRegisterSchemas = true });

var order = new Order { OrderId = "123", Amount = 99.99 };
var eventData = await serializer.SerializeAsync<EventData, Order>(order);

await producerClient.SendAsync(new[] { eventData });

bootstrap.servers=<namespace>.servicebus.windows.net:9093
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
    username="$ConnectionString" \
    password="Endpoint=sb://<namespace>.servicebus.windows.net/;SharedAccessKeyName=...";

# Event Hub name = Kafka topic name
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer

// Uses DefaultAzureCredential — works with Managed Identity, VS, CLI
var credential = new DefaultAzureCredential();
var producerClient = new EventHubProducerClient(
    "<namespace>.servicebus.windows.net",
    "telemetry",
    credential);

var processorClient = new EventProcessorClient(
    checkpointStore,
    "$Default",
    "<namespace>.servicebus.windows.net",
    "telemetry",
    credential);

# Assign Sender role to a Managed Identity
az role assignment create \
  --assignee <principal-id> \
  --role "Azure Event Hubs Data Sender" \
  --scope /subscriptions/<sub>/resourceGroups/<rg>/providers/\
Microsoft.EventHub/namespaces/<namespace>/eventhubs/<eventhub>

# Create private endpoint for Event Hubs namespace
az network private-endpoint create \
  --name myEventHubPrivateEndpoint \
  --resource-group myRG \
  --vnet-name myVNet \
  --subnet mySubnet \
  --private-connection-resource-id /subscriptions/<sub>/resourceGroups/<rg>/providers/\
Microsoft.EventHub/namespaces/<namespace> \
  --group-ids namespace \
  --connection-name myPrivateEndpointConnection

# Create Private DNS Zone for resolution
az network private-dns zone create \
  --resource-group myRG \
  --name privatelink.servicebus.windows.net

{
  "triggers": {
    "When_events_are_available_in_Event_Hub": {
      "type": "ApiConnection",
      "inputs": {
        "host": {
          "connection": {
            "name": "@parameters('$connections')['eventhubs']['connectionId']"
          }
        },
        "method": "get",
        "path": "/@{encodeURIComponent('telemetry')}/events/batch/head",
        "queries": {
          "consumerGroupName": "logicapps-cg",
          "contentType": "application/json",
          "maximumEventsCount": 50
        }
      },
      "recurrence": { "frequency": "Second", "interval": 30 }
    }
  }
}

[FunctionName("ProcessTelemetry")]
public static async Task Run(
    [EventHubTrigger(
        "telemetry",
        Connection = "EventHubConnection",
        ConsumerGroup = "processor-cg")]
    EventData[] events,
    PartitionContext partitionContext,
    ILogger log)
{
    log.LogInformation($"Partition: {partitionContext.PartitionId}, Count: {events.Length}");

    foreach (EventData eventData in events)
    {
        string messageBody = Encoding.UTF8.GetString(eventData.Body.Array,
            eventData.Body.Offset, eventData.Body.Count);
        var telemetry = JsonSerializer.Deserialize<TelemetryEvent>(messageBody);

        log.LogInformation($"Device: {telemetry.DeviceId}, Temp: {telemetry.Temperature}");
        await ProcessTelemetryAsync(telemetry);
    }
}

[FunctionName("ForwardToEventHub")]
[return: EventHub("output-hub", Connection = "EventHubConnection")]
public static string Run(
    [HttpTrigger(AuthorizationLevel.Function, "post")] HttpRequest req,
    ILogger log)
{
    string body = new StreamReader(req.Body).ReadToEnd();
    return body; // Sent directly to output-hub
}

// Multiple events output
[FunctionName("BatchForward")]
public static async Task Run(
    [EventHubTrigger("input-hub", Connection = "EventHubConnection")] EventData[] events,
    [EventHub("output-hub", Connection = "EventHubConnection")] IAsyncCollector<string> outputEvents)
{
    foreach (var e in events)
    {
        var enriched = EnrichEvent(e.Body.ToString());
        await outputEvents.AddAsync(enriched);
    }
}

{
  "version": "2.0",
  "extensions": {
    "eventHubs": {
      "batchCheckpointFrequency": 5,
      "eventProcessorOptions": {
        "maxBatchSize": 256,
        "prefetchCount": 512
      },
      "initialOffsetOptions": {
        "type": "fromStart"
      }
    }
  }
}

-- Real-time aggregation: avg temperature per device per 5-minute tumbling window
SELECT
    DeviceId,
    AVG(Temperature)    AS AvgTemp,
    MAX(Temperature)    AS MaxTemp,
    MIN(Temperature)    AS MinTemp,
    COUNT(*)            AS EventCount,
    System.Timestamp()  AS WindowEnd
INTO [sql-output]
FROM [eventhub-input] TIMESTAMP BY EnqueuedTime
GROUP BY
    DeviceId,
    TumblingWindow(minute, 5)

-- Alert when temperature exceeds threshold
SELECT
    DeviceId,
    Temperature,
    System.Timestamp() AS AlertTime
INTO [servicebus-alerts]
FROM [eventhub-input] TIMESTAMP BY EnqueuedTime
WHERE Temperature > 85.0

AzureMetrics
| where ResourceType == "MICROSOFT.EVENTHUB/NAMESPACES"
| where MetricName == "ThrottledRequests"
| where TimeGenerated > ago(1h)
| summarize TotalThrottled = sum(Total) by bin(TimeGenerated, 5m), Resource
| where TotalThrottled > 0
| order by TimeGenerated desc

AzureDiagnostics
| where ResourceType == "EVENTHUBS"
| where Category == "ArchiveLogs"
| project TimeGenerated, EventHubName_s, PartitionId_s,
          OffsetSequenceNumber_d, LastSequenceNumber_d
| extend Lag = LastSequenceNumber_d - OffsetSequenceNumber_d
| where Lag > 10000
| order by Lag desc

# Pair primary and secondary namespaces
az eventhubs georecovery-alias create \
  --resource-group myRG \
  --namespace-name myPrimaryNamespace \
  --alias myGeoAlias \
  --partner-namespace /subscriptions/<sub>/resourceGroups/<rg>/providers/\
Microsoft.EventHub/namespaces/mySecondaryNamespace

# Producers/consumers always use the alias endpoint:
# myGeoAlias.servicebus.windows.net

# Initiate failover (irreversible — use with care)
az eventhubs georecovery-alias fail-over \
  --resource-group myRG \
  --namespace-name mySecondaryNamespace \
  --alias myGeoAlias