Hubs de Eventos do Azure
Os Hubs de Eventos do Azure são um serviço de ingestão de telemetria de hiperescala que coleta, transforma e armazena milhões de eventos. Enquanto plataforma de transmissão em fluxo distribuída, este serviço garante baixa latência e um tempo de retenção configurável, o que lhe permite introduzir vastas quantidades de telemetria na cloud e ler dados de várias aplicações através de semântica de publicação/subscrição.
Este artigo explica como usar o Streaming Estruturado com Hubs de Eventos do Azure e clusters do Azure Databricks.
Nota
Os Hubs de Eventos do Azure fornecem um ponto de extremidade compatível com o Apache Kafka que você pode usar com o conector Kafka de Streaming Estruturado, disponível no Databricks Runtime, para processar mensagens dos Hubs de Eventos do Azure. O Databricks recomenda usar o conector Kafka de Streaming Estruturado para processar mensagens dos Hubs de Eventos do Azure.
Requisitos
Para obter suporte à versão atual, consulte "Versões mais recentes" no arquivo Leiame do projeto do Azure Event Hubs Spark Connector.
Crie uma biblioteca em seu espaço de trabalho do Azure Databricks usando a coordenada
com.microsoft.azure:azure-eventhubs-spark_2.11:2.3.17Maven .Nota
Este conector é atualizado regularmente e uma versão mais recente pode estar disponível: recomendamos que você puxe o conector mais recente do repositório Maven
Instale a biblioteca criada no cluster.
Esquema
O esquema dos registros é:
| Column | Tipo |
|---|---|
body |
binário |
partition |
string |
offset |
string |
sequenceNumber |
long |
enqueuedTime |
carimbo de data/hora |
publisher |
string |
partitionKey |
string |
properties |
mapa[string,json] |
O body é sempre fornecido como uma matriz de bytes. Use cast("string") para desserializar explicitamente a body coluna.
Guia de Introdução
Vamos começar com um exemplo rápido: WordCount. O bloco de anotações a seguir é tudo o que é necessário para executar o WordCount usando o Streaming Estruturado com Hubs de Eventos do Azure.
Hubs de Eventos do Azure WordCount com bloco de anotações de Streaming Estruturado
Configuração
Esta seção discute as definições de configuração necessárias para trabalhar com Hubs de Eventos.
Para obter orientações detalhadas sobre como configurar o Streaming Estruturado com Hubs de Eventos do Azure, consulte o Guia de Integração de Streaming Estruturado e Hubs de Eventos do Azure desenvolvido pela Microsoft.
Para obter orientações detalhadas sobre como usar o Streaming Estruturado, consulte Streaming no Azure Databricks.
Connection string
Uma cadeia de conexão de Hubs de Eventos é necessária para se conectar ao serviço Hubs de Eventos. Você pode obter a cadeia de conexão para sua instância de Hubs de Eventos no portal do Azure ou usando o ConnectionStringBuilder na biblioteca.
Portal do Azure
Quando você obtém a cadeia de conexão do portal do Azure, ela pode ou não ter a EntityPath chave. Considere:
// Without an entity path
val without = "Endpoint=<endpoint>;SharedAccessKeyName=<key-name>;SharedAccessKey=<key>"
// With an entity path
val with = "Endpoint=sb://<sample>;SharedAccessKeyName=<key-name>;SharedAccessKey=<key>;EntityPath=<eventhub-name>"
Para se conectar aos seus EventHubs, é necessário EntityPath estar presente. Se a sua cadeia de conexão não tiver uma, não se preocupe.
Isso cuidará disso:
import org.apache.spark.eventhubs.ConnectionStringBuilder
val connectionString = ConnectionStringBuilder(without) // defined in the previous code block
.setEventHubName("<eventhub-name>")
.build
ConnectionStringBuilder
Como alternativa, você pode usar o ConnectionStringBuilder para criar sua cadeia de conexão.
import org.apache.spark.eventhubs.ConnectionStringBuilder
val connectionString = ConnectionStringBuilder()
.setNamespaceName("<namespace-name>")
.setEventHubName("<eventhub-name>")
.setSasKeyName("<key-name>")
.setSasKey("<key>")
.build
EventHubsConf
Toda a configuração relacionada aos Hubs de Eventos acontece no seu EventHubsConf. Para criar um EventHubsConf, você deve passar uma cadeia de conexão:
val connectionString = "<event-hub-connection-string>"
val eventHubsConf = EventHubsConf(connectionString)
Consulte Cadeia de conexão para obter mais informações sobre como obter uma cadeia de conexão válida.
Para obter uma lista completa de configurações, consulte EventHubsConf. Aqui está um subconjunto de configurações para você começar:
| Opção | valor | Default | Tipo de consulta | Description |
|---|---|---|---|---|
consumerGroup |
String | "$Default" | Streaming e lote | Um grupo de consumidores é uma visão de um hub de eventos inteiro. Os grupos de consumidores ativam várias aplicações de consumo e cada uma tem uma vista separada do fluxo de eventos e lê o fluxo de forma independente ao seu próprio ritmo e com os seus próprios desvios. Mais informações estão disponíveis na documentação da Microsoft. |
startingPosition |
EventPosition | Início do fluxo | Streaming e lote | A posição inicial para o seu trabalho de Streaming Estruturado. Consulte startingPositions para obter informações sobre a ordem em que as opções são lidas. |
maxEventsPerTrigger |
long | partitionCount * 1000 |
Consulta de streaming | Limite de taxa no número máximo de eventos processados por intervalo de gatilho. O número total especificado de eventos será dividido proporcionalmente em partições de volume diferente. |
Para cada opção, existe uma configuração correspondente em EventHubsConf. Por exemplo:
import org.apache.spark.eventhubs.
val cs = "<your-connection-string>"
val eventHubsConf = EventHubsConf(cs)
.setConsumerGroup("sample-cg")
.setMaxEventsPerTrigger(10000)
EventPosition
EventHubsConf permite que os usuários especifiquem posições iniciais (e finais) com a EventPosition classe. EventPosition define a posição de um evento em uma partição do Hub de Eventos. A posição pode ser um tempo enfileirado, deslocamento, número de sequência, o início do fluxo ou o final do fluxo.
import org.apache.spark.eventhubs._
EventPosition.fromOffset("246812") // Specifies offset 246812
EventPosition.fromSequenceNumber(100L) // Specifies sequence number 100
EventPosition.fromEnqueuedTime(Instant.now) // Any event after the current time
EventPosition.fromStartOfStream // Specifies from start of stream
EventPosition.fromEndOfStream // Specifies from end of stream
Se você gostaria de começar (ou terminar) em uma posição específica, basta criar o correto EventPosition e defini-lo em seu EventHubsConf:
val connectionString = "<event-hub-connection-string>"
val eventHubsConf = EventHubsConf(connectionString)
.setStartingPosition(EventPosition.fromEndOfStream)
Streaming estruturado de produção com Hubs de Eventos do Azure
Quando você executa consultas de streaming em produção, provavelmente deseja mais robustez e garantias de tempo de atividade do que teria quando simplesmente anexa um bloco de anotações a um cluster e executa suas consultas de streaming interativamente. Importe e execute o bloco de anotações a seguir para obter uma demonstração de como configurar e executar o Streaming Estruturado em produção com os Hubs de Eventos do Azure e o Azure Databricks.
Para obter mais informações, consulte Considerações de produção para streaming estruturado.