C3 AI Documentation Home

Azure Event Hubs Connector

The C3 Agentic AI Platform connector for Azure Event Hubs connects C3 Agentic AI Platform applications to your Azure Event Hubs service.

To consume Event Hubs data and persist it in a C3 Agentic AI Platform data store, use the CloudMessageSourceSystem framework. The framework consumes and ingests messages without code from messaging systems such as Azure Event Hubs, AWS Kinesis, and Apache Kafka.

Prerequisites

Before you use the Azure Event Hubs connector, make sure you have access to an active Event Hubs namespace with one or more hubs. See the Azure Event Hubs documentation for more information.

The steps on this page need the C3.AppAdmin or C3.ClusterAdmin role.

Register Azure Event Hubs credentials with the C3 Agentic AI Platform

Define the Event Hub cloud message broker name:

JavaScript
var ehResourceName = "EVENT_HUB_NAME/EVENT_HUB_NAMESPACE/RESOURCE_GROUP";

Generate the configuration key from an AzureEventHub instance that you construct with the broker name. Then set the broker name as the configuration value:

JavaScript
var configKey = AzureEventHub.make({resourceName: ehResourceName}).configKey();
AzureEventHub.setConfigAndSecretValue(configKey, "resourceName", ehResourceName);
AzureEventHub.setConfigAndSecretValue(configKey, "external", true);

Set credentials with a service principal

In this task, you configure authentication for your Azure resources with a service principal. Your application then authenticates without user credentials, which is more secure.

Understand service principal authentication

A service principal is an identity that applications, hosted services, and automated tools use to access Azure resources. A service principal authenticates with a client ID and a client secret.

Gather required information

Before you start, make sure you have the following details:

  1. Active Directory ID: your Microsoft Entra ID tenant ID
  2. Access Key: the application (client) ID of your registered application
  3. Secret Key: the client secret (application password) generated for your application
  4. Account ID: your Azure subscription ID
  5. Region: the Azure region where your resources are located

Create the credentials object

Use the gathered information to create a credentials object in your code, following this template:

JavaScript
var credentials = {
  "type" : "AzureCredentials",
  "activeDirectoryId" : "< ACTIVE_DIRECTORY_ID >",
  "accessKey" : "< APPLICATION_ID >",
  "secretKey" : "< SECRET_ID >",
  "accountId" : "< SUBSCRIPTION_ID >",
  "region" : "< CLOUD_REGION >"
}

Set credentials for your resource

To authenticate your application with Azure resources, call a method such as AzureEventHub.setCredentialsForResourceName and pass in the resource name and the credentials object.

Test your configuration

After you set up the credentials, run your application to confirm that it authenticates and works with Azure resources. Watch for errors and adjust your configuration as needed.

Note on Azure Event Hub client creation

When you provide the following three fields, the platform always tries to create the Azure Event Hub client with the service principal:

  • activeDirectoryId
  • secretKey
  • accessKey

If you populate these fields and also intend to use a shared key, the platform may still create the client with the service principal. That client may have different roles from the ones assigned to the shared key.

Assign roles for Azure Event Hub access

After you set up service principal credentials in Azure, assign the roles that let them work with Azure Event Hub:

  1. Open the Azure portal: sign in to the Azure portal and go to your subscription.

  2. Open access control: select the subscription and go to the Access control (IAM) section.

  3. Add a role assignment:

    a. Select + Add and choose Add role assignment.

    b. Select the Azure Event Hub Owner role to grant full access.

  4. Assign the role to the service principal: search for and select the service principal that needs access.

  5. Verify the assignment: confirm that the service principal appears in the Role assignments section.

With this role, users can send and receive messages through Azure Event Hub.

Test sending and receiving messages in the static console

After you configure service principal authentication for Azure resources, test sending and receiving messages in the static console:

Define Azure Event Hub resource

Set this variable to the name of your Azure Event Hub resource. If you do not have an Azure Event Hub resource, create one first. Make sure the name matches your setup.

JavaScript
eventHubResourceName = "<your_event_hub_resource_name>"

Create a service principal credential

Create a service principal credential that authenticates and authorizes access to Azure resources.

JavaScript
cred=AzureCredentials.make({
  activeDirectoryId: "<your_active_directory_id>",
  secretKey: "<your_client_secret>",
  description: "<description>",
  accountId: "<your_subscription_id>",
  accessKey: "<your_application_client_id>",
  region: "<your_region>",
})

Set the credential

To enable authenticated access to Azure resources, assign the service principal credential to the CloudMessageSourceSystem. Call the setCredentials() method on the CloudMessageSourceSystem instance with the credential you created.

JavaScript
// create an instance of CloudMessageSourceSystem with the name "EventHub" using the make function
cmss=CloudMessageSourceSystem.make({
  "name": "EventHub"
})
//set the credentials
cmss.setCredentials(cred)

Create a cloud message

Create a cloud message to send to the Azure Event Hub with the CloudStreamMessage.make() method:

JavaScript
msg = CloudStreamMessage.make({
  id: "123",
})

The message is now ready to send to the Azure Event Hub.

Send the message

Send the message to the Azure Event Hub with the DataIntegrator.sendObjs() method. The platform delivers the message according to the parameters you pass.

JavaScript
DataIntegrator.sendObjs({'type': 'AzureEventHub', 'resourceName': '<your_event_hub_resource_name>'}, [msg], {'batchSize': 100, 'contentType': "application/json"})

Receive and verify the message

Receive the messages sent to the Azure Event Hub to confirm that the whole process, from sending to receiving, works.

JavaScript
msgs=DataIntegrator.receiveMessages({'type': 'AzureEventHub', 'resourceName': '<your_event_hub_resource_name>'}, {'type': 'CloudStreamReceiveMessagesSpec', 'partition': "1", 'timeoutMillis': 1108, 'batchSize': 100})

Set credentials with an access key

In this task, you configure authentication for Azure Event Hubs with the access key method. This method uses the general Azure credentials setup, which covers several authentication methods.

Understand access key authentication

With the access key method, your application authenticates with Azure resources through a key that grants access to the associated services.

Gather necessary information

Before you continue, make sure you have the following details:

  1. Active Directory ID: your Microsoft Entra ID tenant ID

  2. Access Key: the key used for authentication, either a primary or a secondary key

  3. Secret Key: if applicable, a client secret for additional security

  4. Account ID: your Azure subscription ID

  5. Region: the Azure region where your resources are located

    a. Shared access policy name and key: if you use a shared access signature (SAS), also gather these details.

Create the credentials object

Use the collected information to create a credentials object in your code, following this template:

JavaScript
var credentials = {
  "type": "AzureCredentials",
  "activeDirectoryId": "< ACTIVE_DIRECTORY_ID >", // Tenant ID
  "accessKey": "< ACCESS_KEY >", // Access key
  "secretKey": "< SECRET_ID >", // Optional client secret
  "accountId": "< SUBSCRIPTION_ID >", // Subscription ID
  "region": "< CLOUD_REGION >", // Azure region
  "sharedAccessPolicyName": "< SAS_POLICY_NAME >", // Required for SAS
  "sharedAccessPolicyKey": "< SAS_KEY >" // Required for SAS
}

Set credentials for your resource

To authenticate your application with Azure Event Hubs, call AzureEventHub.setCredentialsForResourceName and pass in the resource name and the credentials object:

JavaScript
AzureEventHub.setCredentialsForResourceName(ehResourceName, credentials, ConfigOverride.APP);

This call applies the credentials that secure the connection to the Event Hub. Your application can then authenticate with the Event Hub and process its events.

Test your configuration

After you set up the credentials, run your application to confirm that it authenticates and works with Azure Event Hubs. Watch for errors and adjust your configuration as needed.

Configure an Azure Event Hub consumer

To subscribe the C3 Agentic AI Platform to an Azure Event Hub, complete these three steps:

  1. Configure the Types to model the messages consumed from the Event Hub.
  2. Configure CloudMessageSourceSystem and CloudMessageSourceCollection seed data.
  3. Enable the CloudMessageSourceCollection to receive messages.

You can also configure a CloudInboundMessage Type to model an Event Hub consumer explicitly. CloudInboundMessage is a parametric Type, so you specify a Type that represents the messages consumed from the Event Hub.

The following example subscribes the C3 Agentic AI Platform to the eh1 Event Hub in Azure Event Hubs.

This example contains messages published to the eh1 hub with the following format:

JSON
{
  "Country": "USA",
  "City": "New York",
  "Date": "2023-05-11",
  "Temperature": 20,
  "Humidity": 60,
  "Wind Speed": 10,
  "Wind Direction": "NW",
  "Precipitation": 0,
  "Cloud Cover": 20,
  "Visibility": 10,
  "Pressure": 1013,
  "Dew Point": 10,
  "UV Index": 5,
  "Sunrise": "05:30",
  "Sunset": "20:15",
  "Moonrise": "22:00",
  "Moonset": "08:00",
  "Moon Phase": "Waning Gibbous",
  "Conditions": "Partly Cloudy",
  "Icon": "partly-cloudy-day"
}

Configure the Types to model the messages consumed from the Event Hub

Define the Types that model the JSON format of the messages you consume from the Event Hub. This example uses one Type, SourceWeatherDataMeasurement, which models the JSON message above.

Type
/**
 * This type represents the raw data that will represent `WeatherDataMeasurement` information.
 */
type SourceWeatherDataMeasurement mixes Source  {

  Country: string

  City: string

  Date: datetime

  Temperature: int

  Humidity: int

  @ser(name="Wind Speed")
  WindSpeed: int

  @ser(name="Wind Direction")
  WindDirection: string

  Precipitation: int

  @ser(name="Cloud Cover")
  CloudCover: int

  Visibility: int

  Pressure: int

  @ser(name="Dew Point")
  DewPoint: int

  @ser(name="UV Index")
  UvIndex: int

  Sunrise: string

  Moonrise: string

  Moonset: string

  @ser(name="Moon Phase")
  MoonPhase: string

  Conditions: string

  Icon: string

}

You may need additional Types, transforms, and Application Types to capture the full models of the messages you consume.

Configure the CloudMessageSourceSystem and CloudMessageSourceCollection Types to register the Event Hub

A CloudMessageSourceSystem organizes related CloudMessageSourceCollection instances.

You create a CloudMessageSourceSystem by specifying its name. For example, add the following EventHub.json file to the \metadata\CloudMessageSourceSystem directory of your package:

JSON
{
    "name": "EventHub"
}

A CloudMessageSourceCollection represents a source collection that references a CloudMessageBroker and its related queue, topic, or stream.

The CloudMessageSourceCollection references a CloudMessageSourceSystem and holds the information needed to process data received from the Event Hub and to connect to the Event Hub.

For example, to model a CloudMessageSourceCollection called WeatherDataMeasurementSource, add the following WeatherDataMeasurementEventHub.json file to the \metadata\CloudMessageSourceCollection directory of your package:

JSON
{
    "name": "WeatherDataMeasurementSource",
    "sourceSystem": {
        "type" : "CloudMessageSourceSystem", 
        "name" : "EventHub"
    },
    "source": "SourceWeatherDataMeasurement",
    "external": true,
    "cloudMessageBrokerName": "eh1/c3-eventhub-testing/azv8qa-rsgp-c3-01",
    "cloudMessageBrokerType": "AzureEventHub",
    "batchSize": 10,
    "timeoutMillis": 5000
}

In the console, you can configure the CloudMessageSourceCollection further with the SourceCollection.Config mixin. The following code sets the content type, the preferences for archiving records, and the SourceCollectionProcessMode.

JavaScript
var config = CloudMessageSourceCollection.forName("WeatherDataMeasurementSource").config();
config.setConfigValue("contentTypeOverride", "application/json", ConfigOverride.APP);
config.setConfigValue("doNotChunk", true, ConfigOverride.APP);
config.setConfigValue("doNotArchiveStatus", false, ConfigOverride.APP);
config.setConfigValue("doNotArchiveSources", true, ConfigOverride.APP);
config.setConfigValue("doNotArchiveFailedSources", false, ConfigOverride.APP);
config.setConfigValue("processMode", "ON_ARRIVAL", ConfigOverride.APP);

Enable the CloudMessageSourceCollection to receive messages

Run the following commands to subscribe the platform to the Event Hub and process messages in real time as they arrive in the broker:

JavaScript
src.register()
src.startOnArrivalMode()

To verify that the platform receives messages, create an AzureEventHub instance from your resource name and call the receiveMessages() method. Include the partition of the Event Hub that you want to read messages from. For more information, see the Partition section of the Features and terminology in Azure Event Hubs topic in the Azure Event Hubs documentation.

JavaScript
broker = AzureEventHub.fromResourceName(ehResourceName);
messages = broker.receiveMessages({partition: 0})

The platform stores messages as a JavaScript ArrayBuffer. Run the following command to read the messages as objects:

JavaScript
Sources.fromCloudMessages(messages, src).readSources().collect();

This command works only when sources are not archived. With archiving on, the platform consumes messages on receipt, so you cannot read them.

Ten weather messages read from the Event Hub

Configure a CloudInboundMessage Type and link to Event Hub

To model an Event Hub consumer and configure downstream processing, create a CloudInboundMessage Type. When a Type mixes in CloudInboundMessage, the platform treats it as an Event Hub consumer.

The platform dispatches each message consumed from the eh1 Event Hub to the SourceWeatherDataMeasurementInboundQueue.process() method.

Type
type SourceWeatherDataMeasurementInboundQueue mixes CloudInboundMessage<SourceWeatherDataMeasurement> {
  
  process: ~ js-server
  
}

Implement the process method in SourceWeatherDataMeasurementInboundQueue.js, which persists data to the WeatherDataMeasurement Type.

JavaScript
function process(messages) {

    var array = WeatherDataMeasurement.array();
    var meas;
  
    messages.forEach(function (msg) {
  
      meas = WeatherDataMeasurement.make({
        country: msg.Country,
        city: msg.City,
        date: msg.Date,
        temperature: msg.Temperature,
        humidity: msg.Humidity,
        windSpeed: msg.WindSpeed,
        windDirection: msg.WindDirection,
        precipitation: msg.Precipitation,
        cloudCover: msg.CloudCover,
        visibility: msg.Visibility,
        pressure: msg.Pressure,
        dewPoint: msg.DewPoint,
        uvIndex: msg.UvIndex,
        sunrise: msg.Sunrise,
        moonrise: msg.Moonrise,
        moonset: msg.Moonset,
        moonPhase: msg.MoonPhase,
        conditions: msg.Conditions,
        icon: msg.Icon
      });
  
      array.push(meas);
  
    })
  
    try {
      WeatherDataMeasurement.createBatch(array);
    } catch (e) {
      log.error(e.message);
    }
  }

The example catches errors from createBatch so that one bad batch does not stop the consumer. The error is logged, and the messages in that batch are not loaded. If you remove the try block, an error in process makes the platform read the same messages again, because delivery is at least once. In that case make process idempotent, for example by using upsertBatch with stable IDs.

Next, associate the SourceWeatherDataMeasurementInboundQueue Type with the eh1 Event Hub.

JavaScript
SourceWeatherDataMeasurementInboundQueue.setBrokerName(ehResourceName, "AzureEventHub");
SourceWeatherDataMeasurementInboundQueue.setBatchSize(10);
SourceWeatherDataMeasurementInboundQueue.setTimeoutMillis(5000);

SourceWeatherDataMeasurementInboundQueue.register();

Data now populates the WeatherDataMeasurement Type.

WeatherDataMeasurement fetch result with count 10 and one record per city, such as Rio de Janeiro and New York

Configure an Azure Event Hubs producer

Publishing to an Event Hub takes nearly the same steps as consuming from one.

Configure a CloudOutboundMessage Type to model an Event Hub producer explicitly. The published messages have the same structure as the consumed ones, so you reuse the SourceWeatherDataMeasurement Type in the CloudOutboundMessage parametric Type.

Configure a CloudOutboundMessage Type and link to Event Hub

When a Type mixes in CloudOutboundMessage, the platform treats it as an Event Hub producer.

Type
type SourceWeatherDataMeasurementOutboundQueue mixes CloudOutboundMessage<SourceWeatherDataMeasurement>

Next, associate the SourceWeatherDataMeasurementOutboundQueue Type with the eh1 Event Hub.

JavaScript
SourceWeatherDataMeasurementOutboundQueue.setBrokerName("eh1/c3-eventhub-testing/azv8qa-rsgp-c3-01", "AzureEventHub");
SourceWeatherDataMeasurementOutboundQueue.setBatchSize(10);

Publish messages to the eh1 Event Hub with the SourceWeatherDataMeasurementOutboundQueue.sendBatch() method. The following code generates sample data and publishes it to the Event Hub:

JavaScript
function createMeasurements(city, country, date) {
  var cAry = C3.ArrayBuilder.fromJson({type: 'ArrayBuilder<SourceWeatherDataMeasurement>'});
  var partition = 1;
  var meas;
  var t = DateTime.fromString(date);

  for (var i=0; i < 10; i++) {
    t = t.plusDays(1);

    meas = SourceWeatherDataMeasurement.make({
        Country: country,
        City: city,
        Date: t,
        Temperature: randomBetween(18, 28),
        Humidity: randomBetween(20, 90),
        WindSpeed: randomBetween(2, 20),
        WindDirection: "SW",
        Precipitation: randomBetween(0, 15),
        CloudCover: randomBetween(0, 10),
        Visibility: randomBetween(25, 50),
        Pressure: randomBetween(900, 1200),
        DewPoint: randomBetween(0, 40),
        UvIndex: randomBetween(0, 10),
        Sunrise: "05:25",
        Moonrise: "21:00",
        Moonset: "09:00",
        MoonPhase: "Waning Gibbous",
        Conditions: "Sunny",
        Icon: "clear-day"
      });

    cAry.push(meas);   
  }

  SourceWeatherDataMeasurementOutboundQueue.sendBatch(cAry.build(), partition);
}

function randomBetween(min,max) {
    return Math.floor(Math.random()*(max-min+1)+min);
}

createMeasurements("Los Angeles", "USA", "2023-06-01");

You can then verify that the messages reached the Event Hub.

Ten weather records for Los Angeles, USA, read back from the Event Hub with their temperature and humidity

See also

Was this page helpful?