Sending Events to Azure Event Hubs

Introduction to Sending Events

Azure Event Hubs is a highly scalable data streaming platform and event ingestion service. Sending events to Event Hubs is a fundamental operation for many real-time data processing scenarios. This guide covers the essential steps and best practices for reliably sending events.

Core Concepts

• Event Producer: An application or service that sends events.
• Event: A unit of data, typically a small piece of information.
• Partition: Events are distributed across partitions within an Event Hub. Understanding partitioning is crucial for ordering and throughput.
• Connection String: Used for authenticating with Event Hubs.

Prerequisites

Before you can send events, ensure you have the following:

• An Azure subscription.
• An Event Hubs namespace created in Azure.
• An Event Hub created within the namespace.
• The connection string for your Event Hub (preferably using a Shared Access Signature with send permissions).

Sending Events using the Azure SDKs

The Azure SDKs provide robust libraries for interacting with Event Hubs. Here are examples for common languages:

using Azure.Messaging.EventHubs;
using System;
using System.Text;
using System.Threading.Tasks;

public class EventSender
{
    private const string EventHubConnectionString = "";
    private const string EventHubName = "";

    public static async Task Main(string[] args)
    {
        await SendEventAsync("Hello from C#! This is event #1.");
        await SendEventAsync("Another event from C#.");
    }

    public static async Task SendEventAsync(string eventData)
    {
        await using var producerClient = new EventDataSender(EventHubConnectionString, EventHubName);
        try
        {
            var eventBody = new EventData(Encoding.UTF8.GetBytes(eventData));
            Console.WriteLine($"Sending event: {eventData}");
            await producerClient.SendAsync(eventBody);
            Console.WriteLine("Event sent successfully.");
        }
        catch (Exception ex)
        {
            Console.WriteLine($"Error sending event: {ex.Message}");
        }
    }
}
                

from azure.eventhub import EventHubProducerClient
import os

connection_str = os.environ["EVENTHUB_CONNECTION_STR"]
eventhub_name = os.environ["EVENTHUB_NAME"]

producer_client = EventHubProducerClient.from_connection_string(connection_str, eventhub_name)

async def send_event(event_data):
    async with producer_client:
        event_data_batch = await producer_client.create_batch()
        event_data_batch.add(event_data.encode('utf-8'))
        await producer_client.send_batch(event_data_batch)
        print(f"Sent event: {event_data}")

async def main():
    await send_event("Hello from Python! This is event #1.")
    await send_event("Another event from Python.")

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())
                

import com.azure.messaging.eventhubs.EventData;
import com.azure.messaging.eventhubs.EventHubProducerClient;
import com.azure.messaging.eventhubs.EventHubProducerClientBuilder;
import java.util.Arrays;
import java.util.List;

public class EventSender {

    private static final String EVENT_HUB_CONNECTION_STRING = "";
    private static final String EVENT_HUB_NAME = "";

    public static void main(String[] args) {
        sendEvent("Hello from Java! This is event #1.");
        sendEvent("Another event from Java.");
    }

    public static void sendEvent(String eventData) {
        EventHubProducerClient producerClient = new EventHubProducerClientBuilder()
            .connectionString(EVENT_HUB_CONNECTION_STRING)
            .hubName(EVENT_HUB_NAME)
            .buildClient();

        try {
            List<EventData> events = Arrays.asList(new EventData(eventData.getBytes()));
            System.out.println("Sending event: " + eventData);
            producerClient.send(events);
            System.out.println("Event sent successfully.");
        } catch (Exception e) {
            System.err.println("Error sending event: " + e.getMessage());
        } finally {
            producerClient.close();
        }
    }
}
                

const { EventHubProducerClient } = require("@azure/event-hubs");

const connectionString = process.env.EVENTHUB_CONNECTION_STR;
const eventHubName = process.env.EVENTHUB_NAME;

const producerClient = new EventHubProducerClient(connectionString, eventHubName);

async function sendEvent(eventData) {
    try {
        const batch = await producerClient.createBatch();
        batch.add({ body: eventData });
        await producerClient.sendBatch(batch);
        console.log(`Sent event: ${eventData}`);
    } catch (err) {
        console.error("Error sending event:", err);
    }
}

async function main() {
    await sendEvent("Hello from Node.js! This is event #1.");
    await sendEvent("Another event from Node.js.");
    await producerClient.close();
}

main().catch(error => {
    console.error("An unexpected error occurred:", error);
    process.exit(1);
});
                

Best Practices for Sending Events

• Batching: Send events in batches whenever possible to improve throughput and reduce costs. The SDKs provide mechanisms for creating and sending batches.
• Partitioning Strategy: If you need strict ordering for a set of related events, use a partition key. All events with the same partition key will be sent to the same partition.
• Error Handling: Implement robust error handling and retry mechanisms. Network issues or transient service errors can occur.
• Asynchronous Operations: Utilize asynchronous programming patterns to avoid blocking your application threads, especially in high-throughput scenarios.
• Connection Management: Reuse producer clients for multiple send operations to minimize overhead. Close clients when they are no longer needed.
• Large Events: For very large events, consider breaking them down or using Event Hubs' large message support if available and appropriate for your scenario.

Advanced Scenarios

Sending Events with Custom Properties

You can attach custom metadata to your events. This is useful for adding contextual information.

// Example in C#
var eventBody = new EventData(Encoding.UTF8.GetBytes("My event data"));
eventBody.Properties.Add("CorrelationId", Guid.NewGuid().ToString());
eventBody.Properties.Add("SourceSystem", "MyApp");
await producerClient.SendAsync(eventBody);
            

Using Partition Keys

To ensure events are sent to the same partition:

// Example in C#
var eventBody = new EventData(Encoding.UTF8.GetBytes("User login event"));
await producerClient.SendAsync(eventBody, new SendEventOptions { PartitionKey = "user-123" });