Configurar a cache de dados e a persistência em disco para fluxos de dados

Quando um fluxo de dados envia mensagens para um ponto final de destino, como Hubs de Eventos do Azure, Microsoft Fabric Real-Time Intelligence, Kafka ou outro serviço cloud, o destino ou a rede podem tornar-se indisponíveis. O Operações IoT do Azure utiliza o modelo de subscrição local do corretor MQTT para armazenar mensagens e tentar a entrega.

Este artigo explica como os fluxos de dados protegem as mensagens durante falhas de destino e como configurar o buffering do broker e a persistência do disco para uma proteção mais forte.

Como os dados fluem em buffer as mensagens durante as interrupções de destino

Quando usa o corretor MQTT local como ponto final de origem num fluxo de dados, seja direta ou indiretamente através de ativos que publicam nele, o fluxo de dados recebe mensagens do corretor MQTT como assinante. O fluxo de dados reconhece cada mensagem de origem depois de a mensagem ser processada com sucesso e entregue ao destino, ou depois de a mensagem ser intencionalmente descartada devido a filtragem, validação de esquema ou expiração da mensagem.

Se o destino final não estiver disponível, a entrega não pode ser concluída. Neste caso, o fluxo de dados não reconhece a mensagem de origem. O corretor MQTT mantém a mensagem na fila de assinantes e o fluxo de dados reitera a entrega. Quando a conectividade é restabelecida, o fluxo de dados envia mensagens em fila para o destino e reconhece-as após a entrega bem-sucedida.

O caminho da mensagem é: publicador ou ativo para o corretor MQTT, intermediário MQTT para fluxo de dados e fluxo de dados para o endpoint de destino.

1. O corretor MQTT entrega uma mensagem ao fluxo de dados.
2. O fluxo de dados envia a mensagem para o ponto final de destino.
3. Se o envio for bem-sucedido, o fluxo de dados reconhece a mensagem de origem e o corretor remove-a da fila dos assinantes.
4. Se o envio falhar, o fluxo de dados não reconhece a mensagem de origem. O corretor mantém-no em fila.
5. O fluxo de dados tenta a entrega novamente até que seja bem-sucedida, a mensagem expire ou se aplique um limite configurado.

Camadas de configuração de proteção de dados

Utilize as seguintes camadas de configuração para controlar como as mensagens são armazenadas e protegidas durante as falhas de destino.

A fila de subscritores local do corretor MQTT é armazenada na memória por defeito. Pode configurar o corretor MQTT para usar disco de duas formas diferentes:

• Buffer de mensagens com suporte por disco: Utiliza o disco como buffer de transbordo quando as filas crescem para além da memória disponível. Esta configuração ajuda a lidar com maiores atrasos temporários, mas não é o mesmo que uma persistência duradoura entre reinicializações dos brokers.
• Persistência do corretor MQTT: Persiste dados selecionados do corredor, incluindo filas de assinantes, no disco para que as mensagens em fila possam sobreviver a reinicios ou falhas de energia.

Para detalhes de configuração do corretor, veja:

• Configure as definições do broker para alta disponibilidade, escalabilidade e uso de memória
• Configurar o comportamento do buffer de mensagens apoiado em disco
• Configurar a persistência do MQTT broker
• Configurar as opções do cliente MQTT do broker

Escolha uma configuração de buffering

Escolha uma configuração com base na duração da interrupção e nos requisitos de durabilidade para a sua carga de trabalho:

• Para interrupções curtas e transitórias na cloud ou na rede, a fila de assinantes em memória padrão pode ser suficiente.
• Para maior capacidade ou interrupções temporárias mais longas, configure o buffer de mensagens com suporte em disco.
• Para proteção contra reinício ou perda de energia, ative a persistência do MQTT broker e a persistência da fila de assinantes, depois ative requestDiskPersistence no fluxo de dados ou no gráfico de fluxo de dados.
• Para ambientes de armazenamento limitado, configure limites de fila de assinantes, expiração de mensagens e monitorização para que o corretor aplique limites de fila e elimine ou rejeite mensagens de acordo com a sua política.

Exemplo: Interrupção no destino

Assuma que cria um fluxo de dados usando o broker local MQTT predefinido como endpoint de origem e Hubs de Eventos do Azure como endpoint de destino. Se a ligação entre o fluxo de dados e o Hubs de Eventos do Azure for perdida, o fluxo de dados tenta enviar novamente e não reconhece as mensagens de origem. O corretor do MQTT coloca as mensagens não confirmadas na fila. Com as definições padrão, a fila é armazenada na memória. Com o buffer de mensagens apoiado por disco, a fila pode ser transferida para o disco. Com persistência do broker e fluxo de dados requestDiskPersistence, mensagens em fila podem sobreviver a reinicializações do broker, sujeitas aos limites configurados de persistência, validade e armazenamento.

Ativar a persistência do disco para um fluxo de dados

A persistência do disco permite que os fluxos de dados e os gráficos de fluxo de dados mantenham o estado de processamento durante os reinícios. Quando ativa esta funcionalidade, o corretor MQTT mantém dados, como mensagens na fila de assinantes, no disco. Esta abordagem ajuda a garantir que a fonte de dados do seu fluxo de dados não perca dados em espera durante cortes de energia ou reinícios do broker. O corretor mantém o desempenho ótimo porque a persistência é configurada por fluxo de dados, pelo que apenas os fluxos de dados que dela necessitam utilizam esta funcionalidade.

O fluxo de dados exige persistência durante a subscrição utilizando uma propriedade de utilizador MQTTv5. Esta funcionalidade funciona apenas quando:

• O fluxo de dados utiliza como fonte o corretor ou recurso MQTT.
• O corretor MQTT tem a persistência ativada, com o modo de persistência dinâmica definido para o tipo de dado Enabled, como filas de assinantes.

Para obter detalhes sobre a configuração de persistência do MQTT broker, consulte Configure MQTT broker persistence.

A configuração aceita Enabled ou Disabled. Disabled é o padrão.

Configurar um fluxo de dados

{
    "mode": "Enabled",
    "requestDiskPersistence": "Enabled",
    "operations": [
        // ... your data flow operations
    ]
}

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2025-10-01' = {
  parent: defaultDataflowProfile
  name: dataflowName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    requestDiskPersistence: 'Enabled'
    operations: [
      // ... your data flow operations
    ]
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: <DATAFLOW_NAME>
  namespace: azure-iot-operations
spec:
  profileRef: default 
  mode: Enabled
  requestDiskPersistence: Enabled
  operations:
    # ... your data flow operations

Configurar um grafo de fluxo de dados

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  parent: dataflowProfile
  name: 'my-graph'
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    requestDiskPersistence: 'Enabled'
    nodes: [
      // ... your graph nodes
    ]
    nodeConnections: [
      // ... your node connections
    ]
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: <GRAPH_NAME>
  namespace: azure-iot-operations
spec:
  profileRef: default
  requestDiskPersistence: Enabled
  nodes:
    # ... your graph nodes
  nodeConnections:
    # ... your node connections

Conteúdo relacionado

• Criar um fluxo de dados
• Visão geral dos gráficos de fluxo de dados
• Configurar a persistência do MQTT broker
• Configurar o comportamento do buffer de mensagens apoiado em disco