Use Event Hubs and .NET to send and receive Atlas Kafka topics messages

This quickstart teaches you how to send and receive Atlas Kafka topics events. We'll make use of Azure Event Hubs and the Azure.Messaging.EventHubs .NET library.

Important

A managed event hub is created automatically when your Microsoft Purview account is created. See, Purview account creation. You can publish messages to Event Hubs Kafka topic, ATLAS_HOOK. Purview will receive it, process it and notify Kafka topic ATLAS_ENTITIES of entity changes. This quickstart uses the new Azure.Messaging.EventHubs library.

Prerequisites

If you're new to Event Hubs, see Event Hubs overview before you complete this quickstart.

To follow this quickstart, you need certain prerequisites in place:

  • A Microsoft Azure subscription. To use Azure services, including Event Hubs, you need an Azure subscription. If you don't have an Azure account, you can sign up for a free trial or use your MSDN subscriber benefits when you create an account.

  • Microsoft Visual Studio 2022. The Event Hubs client library makes use of new features that were introduced in C# 8.0. You can still use the library with previous C# versions, but the new syntax won't be available. To make use of the full syntax, it's recommended that you compile with the .NET Core SDK 3.0 or higher and language version set to latest. If you're using a Visual Studio version prior to Visual Studio 2019, it doesn't have the tools needed to build C# 8.0 projects. Visual Studio 2022, including the free Community edition, can be downloaded here.

  • An active Microsoft Purview account with an Event Hubs namespace enabled. This option can be enabled during account creation or in Managed Resources under settings on your Microsoft Purview account page in the Azure portal. Select the toggle to enable, then save. It can take a few minutes for the namespace to be available after it's enabled. Screenshot showing the Event Hubs namespace toggle highlighted on the Managed resources page of the Microsoft Purview account page in the Azure portal.

    Note

    Enabling this Event Hubs namespace does incur a cost for the namespace. For specific details, see the pricing page.

Publish messages to Purview

Let's create a .NET Core console application that sends events to Purview via Event Hubs Kafka topic, ATLAS_HOOK.

Create a Visual Studio project

Next create a C# .NET console application in Visual Studio:

  1. Launch Visual Studio.
  2. In the Start window, select Create a new project > Console App (.NET Framework). .NET version 4.5.2 or above is required.
  3. In Project name, enter PurviewKafkaProducer.
  4. Select Create to create the project.

Create a console application

  1. Start Visual Studio 2022.
  2. Select Create a new project.
  3. On the Create a new project dialog box, do the following steps: If you don't see this dialog box, select File on the menu, select New, and then select Project.
    1. Select C# for the programming language.
    2. Select Console for the type of the application.
    3. Select Console App (.NET Core) from the results list.
    4. Then, select Next.

Add the Event Hubs NuGet package

  1. Select Tools > NuGet Package Manager > Package Manager Console from the menu.

  2. Run the following command to install the Azure.Messaging.EventHubs NuGet package and Azure.Messaging.EventHubs.Producer NuGet package:

    Install-Package Azure.Messaging.EventHubs
    
    Install-Package Azure.Messaging.EventHubs.Producer
    

Write code that sends messages to the event hub

  1. Add the following using statements to the top of the Program.cs file:

    using System;
    using System.Text;
    using System.Threading.Tasks;
    using Azure.Messaging.EventHubs;
    using Azure.Messaging.EventHubs.Producer;
    
  2. Add constants to the Program class for the Event Hubs connection string and Event Hubs name.

    private const string connectionString = "<EVENT HUBS NAMESPACE - CONNECTION STRING>";
    private const string eventHubName = "<EVENT HUB NAME>";
    

    You can get the Event Hubs namespace associated with the Purview account by looking at the Atlas kafka endpoint primary/secondary connection strings. These can be found in Properties tab of your Purview account.

    A screenshot that shows an Event Hubs Namespace.

    The event hub name for sending messages to Purview is ATLAS_HOOK.

  3. Replace the Main method with the following async Main method and add an async ProduceMessage to push messages into Purview. See the comments in the code for details.

        static async Task Main()
        {
            // Read from the default consumer group: $Default
            string consumerGroup = EventHubConsumerClient.DefaultConsumerGroupName;
    
     		/ Create an event producer client to add events in the event hub
            EventHubProducerClient producer = new EventHubProducerClient(ehubNamespaceConnectionString, eventHubName);
    
     		await ProduceMessage(producer);
        }
    
     	static async Task ProduceMessage(EventHubProducerClient producer)
    
        {
     		// Create a batch of events 
     		using EventDataBatch eventBatch = await producerClient.CreateBatchAsync();
    
     		// Add events to the batch. An event is a represented by a collection of bytes and metadata. 
     		eventBatch.TryAdd(new EventData(Encoding.UTF8.GetBytes("<First event>")));
     		eventBatch.TryAdd(new EventData(Encoding.UTF8.GetBytes("<Second event>")));
     		eventBatch.TryAdd(new EventData(Encoding.UTF8.GetBytes("<Third event>")));
    
     		// Use the producer client to send the batch of events to the event hub
     		await producerClient.SendAsync(eventBatch);
     		Console.WriteLine("A batch of 3 events has been published.");
    
     	}
    
  4. Build the project. Ensure that there are no errors.

  5. Run the program and wait for the confirmation message.

    Note

    For the complete source code with more informational comments, see this file in GitHub

Sample code that creates an sql table with two columns using a Create Entity JSON message

	
	{
    "msgCreatedBy":"nayenama",
    "message":{
        "type":"ENTITY_CREATE_V2",
        "user":"admin",
        "entities":{
            "entities":[
                {
                    "typeName":"azure_sql_table",
                    "attributes":{
                        "owner":"admin",
                        "temporary":false,
                        "qualifiedName":"mssql://nayenamakafka.eventhub.sql.net/salespool/dbo/SalesOrderTable",
                        "name":"SalesOrderTable",
                        "description":"Sales Order Table added via Kafka"
                    },
                    "relationshipAttributes":{
                        "columns":[
                            {
                                "guid":"-1102395743156037",
                                "typeName":"azure_sql_column",
                                "uniqueAttributes":{
                                    "qualifiedName":"mssql://nayenamakafka.eventhub.sql.net/salespool/dbo/SalesOrderTable#OrderID"
                                }
                            },
                            {
                                "guid":"-1102395743156038",
                                "typeName":"azure_sql_column",
                                "uniqueAttributes":{
                                    "qualifiedName":"mssql://nayenamakafka.eventhub.sql.net/salespool/dbo/SalesOrderTable#OrderDate"
                                }
                            }
                        ]
                    },
                    "guid":"-1102395743156036",
                    "version":0
                }
            ],
            "referredEntities":{
                "-1102395743156037":{
                    "typeName":"azure_sql_column",
                    "attributes":{
                        "owner":null,
                        "userTypeId":61,
                        "qualifiedName":"mssql://nayenamakafka.eventhub.sql.net/salespool/dbo/SalesOrderTable#OrderID",
                        "precision":23,
                        "length":8,
                        "description":"Sales Order ID",
                        "scale":3,
                        "name":"OrderID",
                        "data_type":"int"
                    },
                    "relationshipAttributes":{
                        "table":{
                            "guid":"-1102395743156036",
                            "typeName":"azure_sql_table",
                            "entityStatus":"ACTIVE",
                            "displayText":"SalesOrderTable",
                            "uniqueAttributes":{
                                "qualifiedName":"mssql://nayenamakafka.eventhub.sql.net/salespool/dbo/SalesOrderTable"
                            }
                        }
                    },
                    "guid":"-1102395743156037",
                    "version":2
                },
                "-1102395743156038":{
                    "typeName":"azure_sql_column",
                    "attributes":{
                        "owner":null,
                        "userTypeId":61,
                        "qualifiedName":"mssql://nayenamakafka.eventhub.sql.net/salespool/dbo/SalesOrderTable#OrderDate",
                        "description":"Sales Order Date",
                        "scale":3,
                        "name":"OrderDate",
                        "data_type":"datetime"
                    },
                    "relationshipAttributes":{
                        "table":{
                            "guid":"-1102395743156036",
                            "typeName":"azure_sql_table",
                            "entityStatus":"ACTIVE",
                            "displayText":"SalesOrderTable",
                            "uniqueAttributes":{
                                "qualifiedName":"mssql://nayenamakafka.eventhub.sql.net/salespool/dbo/SalesOrderTable"
                            }
                        }
                    },
                    "guid":"-1102395743156038",
                    "status":"ACTIVE",
                    "createdBy":"ServiceAdmin",
                    "version":0
                }
            }
        }
    },
    "version":{
        "version":"1.0.0"
    },
    "msgCompressionKind":"NONE",
    "msgSplitIdx":1,
    "msgSplitCount":1
}


Receive Purview messages

Next learn how to write a .NET Core console application that receives messages from event hubs using an event processor. The event processor manages persistent checkpoints and parallel receptions from event hubs. This simplifies the process of receiving events. You need to use the ATLAS_ENTITIES event hub to receive messages from Purview.

Warning

Event Hubs SDK uses the most recent version of Storage API available. That version may not necessarily be available on your Stack Hub platform. If you run this code on Azure Stack Hub, you will experience runtime errors unless you target the specific version you are using. If you're using Azure Blob Storage as a checkpoint store, review the supported Azure Storage API version for your Azure Stack Hub build and in your code, target that version.

The highest available version of the Storage service is version 2019-02-02. By default, the Event Hubs SDK client library uses the highest available version on Azure (2019-07-07 at the time of the release of the SDK). If you are using Azure Stack Hub version 2005, in addition to following the steps in this section, you will also need to add code that targets the Storage service API version 2019-02-02. To learn how to target a specific Storage API version, see this sample in GitHub.

Create an Azure Storage and a blob container

We'll use Azure Storage as the checkpoint store. Use the following steps to create an Azure Storage account.

  1. Create an Azure Storage account

  2. Create a blob container

  3. Get the connection string for the storage account

    Make note of the connection string and the container name. You'll use them in the receive code.

Create a Visual Studio project for the receiver

  1. In the Solution Explorer window, select and hold (or right-click) the EventHubQuickStart solution, point to Add, and select New Project.
  2. Select Console App (.NET Core), and select Next.
  3. Enter PurviewKafkaConsumer for the Project name, and select Create.

Add the Event Hubs NuGet package

  1. Select Tools > NuGet Package Manager > Package Manager Console from the menu.

  2. Run the following command to install the Azure.Messaging.EventHubs NuGet package:

    Install-Package Azure.Messaging.EventHubs
    
  3. Run the following command to install the Azure.Messaging.EventHubs.Processor NuGet package:

    Install-Package Azure.Messaging.EventHubs.Processor
    

Update the Main method

  1. Add the following using statements at the top of the Program.cs file.

    using System;
    using System.Text;
    using System.Threading.Tasks;
    using Azure.Storage.Blobs;
    using Azure.Messaging.EventHubs;
    using Azure.Messaging.EventHubs.Consumer;
    using Azure.Messaging.EventHubs.Processor;
    
  2. Add constants to the Program class for the Event Hubs connection string and the event hub name. Replace placeholders in brackets with the real values that you got when you created the event hub and the storage account (access keys - primary connection string). Make sure that the {Event Hubs namespace connection string} is the namespace-level connection string, and not the event hub string.

        private const string ehubNamespaceConnectionString = "<EVENT HUBS NAMESPACE - CONNECTION STRING>";
        private const string eventHubName = "<EVENT HUB NAME>";
        private const string blobStorageConnectionString = "<AZURE STORAGE CONNECTION STRING>";
        private const string blobContainerName = "<BLOB CONTAINER NAME>";
    

    You can get event hub namespace associated with your Purview account by looking at your Atlas kafka endpoint primary/secondary connection strings. This can be found in the Properties tab of your Purview account.

    A screenshot that shows an Event Hubs Namespace.

    Use ATLAS_ENTITIES as the event hub name when sending messages to Purview.

  3. Replace the Main method with the following async Main method. See the comments in the code for details.

        static async Task Main()
        {
            // Read from the default consumer group: $Default
            string consumerGroup = EventHubConsumerClient.DefaultConsumerGroupName;
    
            // Create a blob container client that the event processor will use 
            BlobContainerClient storageClient = new BlobContainerClient(blobStorageConnectionString, blobContainerName);
    
            // Create an event processor client to process events in the event hub
            EventProcessorClient processor = new EventProcessorClient(storageClient, consumerGroup, ehubNamespaceConnectionString, eventHubName);
    
            // Register handlers for processing events and handling errors
            processor.ProcessEventAsync += ProcessEventHandler;
            processor.ProcessErrorAsync += ProcessErrorHandler;
    
            // Start the processing
            await processor.StartProcessingAsync();
    
            // Wait for 10 seconds for the events to be processed
            await Task.Delay(TimeSpan.FromSeconds(10));
    
            // Stop the processing
            await processor.StopProcessingAsync();
        }    
    
  4. Now add the following event and error handler methods to the class.

        static async Task ProcessEventHandler(ProcessEventArgs eventArgs)
        {
            // Write the body of the event to the console window
            Console.WriteLine("\tReceived event: {0}", Encoding.UTF8.GetString(eventArgs.Data.Body.ToArray()));
    
            // Update checkpoint in the blob storage so that the app receives only new events the next time it's run
            await eventArgs.UpdateCheckpointAsync(eventArgs.CancellationToken);
        }
    
        static Task ProcessErrorHandler(ProcessErrorEventArgs eventArgs)
        {
            // Write details about the error to the console window
            Console.WriteLine($"\tPartition '{ eventArgs.PartitionId}': an unhandled exception was encountered. This was not expected to happen.");
            Console.WriteLine(eventArgs.Exception.Message);
            return Task.CompletedTask;
        }    
    
  5. Build the project. Ensure that there are no errors.

    Note

    For the complete source code with more informational comments, see this file on the GitHub.

  6. Run the receiver application.

An example of a Message received from Purview

{
	"version":
		{"version":"1.0.0",
		 "versionParts":[1]
		},
		 "msgCompressionKind":"NONE",
		 "msgSplitIdx":1,
		 "msgSplitCount":1,
		 "msgSourceIP":"10.244.155.5",
		 "msgCreatedBy":
		 "",
		 "msgCreationTime":1618588940869,
		 "message":{
			"type":"ENTITY_NOTIFICATION_V2",
			"entity":{
				"typeName":"azure_sql_table",
					"attributes":{
						"owner":"admin",
						"createTime":0,
						"qualifiedName":"mssql://nayenamakafka.eventhub.sql.net/salespool/dbo/SalesOrderTable",
						"name":"SalesOrderTable",
						"description":"Sales Order Table"
						},
						"guid":"ead5abc7-00a4-4d81-8432-d5f6f6f60000",
						"status":"ACTIVE",
						"displayText":"SalesOrderTable"
					},
					"operationType":"ENTITY_UPDATE",
					"eventTime":1618588940567
				}
}

Next steps

Check out more examples in GitHub.