Usar os Hubs de Eventos e o .NET para enviar e receber mensagens de tópicos do Atlas Kafka

Este guia de início rápido ensina como enviar e receber eventos de tópicos do Atlas Kafka . Usaremos os Hubs de Eventos do Azure e o Hub de Eventos do Azure. Biblioteca .NET Messaging.EventHubs.

Pré-requisitos

Se você não estiver familiarizado com os Hubs de Eventos, consulte a visão geral dos Hubs de Eventos antes de concluir este início rápido.

Para seguir este início rápido, você precisa de determinados pré-requisitos:

• Uma assinatura do Microsoft Azure. Para usar os serviços do Azure, incluindo Hubs de Eventos, você precisa de uma assinatura do Azure. Se você não tiver uma conta do Azure, poderá se inscrever para uma avaliação gratuita ou usar os benefícios de assinante do MSDN ao criar uma conta.
• Microsoft Visual Studio 2022. A biblioteca de cliente dos Hubs de Eventos usa novos recursos que foram introduzidos no C# 8.0. Você ainda pode usar a biblioteca com versões anteriores do C#, mas a nova sintaxe não estará disponível. Para usar a sintaxe completa, é recomendável compilar com o SDK do .NET Core 3.0 ou superior e a versão da linguagem definida como latest. Se você estiver usando uma versão do Visual Studio anterior ao Visual Studio 2019, ele não terá as ferramentas necessárias para criar projetos C# 8.0. O Visual Studio 2022, incluindo a edição gratuita da Comunidade, pode ser baixado aqui.
• Uma conta ativa do Microsoft Purview.
• Um Hubs de Eventos configurado com sua conta do Microsoft Purview para enviar e receber mensagens:
  • Sua conta já pode estar configurada. Você pode marcar sua conta do Microsoft Purview no portal do Azure em Configurações, Configuração do Kafka. Se ainda não estiver configurado, siga este guia.

Publicar mensagens no Microsoft Purview

Vamos criar um aplicativo de console .NET Core que envie eventos para o Microsoft Purview por meio do tópico Kafka de Hubs de Eventos, ATLAS_HOOK.

Para publicar mensagens no Microsoft Purview, você precisa de um Hubs de Eventos gerenciado ou pelo menos um Hubs de Eventos com uma configuração de gancho.

Criar um projeto do Visual Studio

Em seguida, crie um aplicativo de console C# .NET no Visual Studio:

1. Inicie o Visual Studio.
2. Na janela Iniciar, selecione Criar um novo projeto>Aplicativo de Console (.NET Framework). O .NET versão 4.5.2 ou superior é necessário.
3. Em Nome do projeto, insira PurviewKafkaProducer.
4. Selecione Criar para criar o projeto.

Criar um aplicativo de console

1. Inicie o Visual Studio 2022.
2. Selecione Criar um novo projeto.
3. Na caixa de diálogo Criar um novo projeto , execute as seguintes etapas: Se você não vir essa caixa de diálogo, selecione Arquivo no menu, selecione Novo e, em seguida, selecione Projeto.
   1. Selecione C# para a linguagem de programação.
   2. Selecione Console para o tipo de aplicativo.
   3. Selecione o Aplicativo de Console (.NET Core) na lista de resultados.
   4. Em seguida, selecione Avançar.

Adicionar o pacote NuGet dos Hubs de Eventos

1. Selecione Ferramentas>Gerenciador>de Pacotes NuGetConsole do Gerenciador de Pacotes no menu.

2. Execute o seguinte comando para instalar o Azure. Messaging.EventHubs, pacote NuGet e o Azure. Pacote NuGet Messaging.EventHubs.Producer:

   Install-Package Azure.Messaging.EventHubs
   

   Install-Package Azure.Messaging.EventHubs.Producer
   

Gravar código que envia mensagens para o hub de eventos

1. Adicione as seguintes using instruções à parte superior do arquivo Program.cs :

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

2. Adicione constantes à Program classe para a cadeia de conexão dos Hubs de Eventos e o nome dos Hubs de Eventos.

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

3. Substitua o Main método pelo método a seguir async Main e adicione um async ProduceMessage para enviar mensagens por push para o Microsoft Purview. Consulte os comentários no código para obter detalhes.

       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. Compile o projeto. Assegure-se de que não existem erros.

5. Execute o programa e aguarde a mensagem de confirmação.

   Observação

   Para obter o código-fonte completo com mais comentários informativos, consulte este arquivo no GitHub

Código de exemplo que cria uma tabela SQL com duas colunas usando uma mensagem JSON Criar Entidade

	
	{
    "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
}

Receber mensagens do Microsoft Purview

Em seguida, aprenda a escrever um aplicativo de console .NET Core que receba mensagens de hubs de eventos usando um processador de eventos. O processador de eventos gerencia pontos de verificação persistentes e recepções paralelas de hubs de eventos. Isso simplifica o processo de recebimento de eventos. Você precisa usar o hub de eventos ATLAS_ENTITIES para receber mensagens do Microsoft Purview.

Para receber mensagens do Microsoft Purview, você precisa de uma configuração de notificação de Hubs de Eventos gerenciados ou de Hubs de Eventos.

Criar um Armazenamento do Azure e um contêiner de blob

Usaremos o Armazenamento do Azure como o repositório de ponto de verificação. Use as etapas a seguir para criar uma conta de Armazenamento do Azure.

1. Configurar uma conta de armazenamento do Azure

2. Criar um contêiner de blobs

3. Obter a cadeia de cadeia de conexão para a conta de armazenamento

   Anote a cadeia de cadeia de conexão e o nome do contêiner. Você os usará no código de recebimento.

Criar um projeto do Visual Studio para o receptor

1. Na janela do Gerenciador de Soluções, selecione e segure (ou clique com o botão direito do mouse) a solução EventHubQuickStart, aponte para Adicionar e selecione Novo Projeto.
2. Selecione Aplicativo de Console (.NET Core) e selecione Avançar.
3. Insira PurviewKafkaConsumer como o nome do Projeto e selecione Criar.

Adicionar o pacote NuGet dos Hubs de Eventos

1. Selecione Ferramentas>Gerenciador>de Pacotes NuGetConsole do Gerenciador de Pacotes no menu.

2. Execute o seguinte comando para instalar o Azure. Pacote NuGet Messaging.EventHubs:

   Install-Package Azure.Messaging.EventHubs
   

3. Execute o seguinte comando para instalar o Azure. Pacote NuGet Messaging.EventHubs.Processor:

   Install-Package Azure.Messaging.EventHubs.Processor
   

Atualizar o método Main

1. Adicione as instruções a seguir using na parte superior do arquivo Program.cs .

   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. Adicione constantes à Program classe para a cadeia de cadeia de conexão dos Hubs de Eventos e o nome do hub de eventos. Substitua os espaços reservados entre colchetes pelos valores reais obtidos ao criar o hub de eventos e a conta de armazenamento (chaves de acesso - cadeia de conexão primária). Certifique-se de que é {Event Hubs namespace connection string} a cadeia de cadeia de conexão no nível do namespace e não a cadeia de caracteres do hub de eventos.

       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>";
   

   Use ATLAS_ENTITIES como nome do hub de eventos ao enviar mensagens para o Microsoft Purview.

3. Substitua o Main método pelo método a seguir async Main . Consulte os comentários no código para obter detalhes.

       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. Agora, adicione os seguintes métodos de manipulador de eventos e erros à classe.

       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. Compile o projeto. Assegure-se de que não existem erros.

   Observação

   Para obter o código-fonte completo com comentários mais informativos, consulte este arquivo no GitHub.

6. Execute o aplicativo receptor.

Um exemplo de uma mensagem recebida do Microsoft 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
				}
}

Próximas etapas

Confira mais exemplos no GitHub.

• Exemplos de Hubs de Eventos no GitHub
• Exemplos de processador de eventos no GitHub
• Uma introdução às notificações do Atlas