Limitar dados em gráficos de fluxo de dados

Uma transformada de aceleração limita a frequência com que encaminha mensagens sobre um tópico MQTT. Em vez de eliminar mensagens com base no seu conteúdo, a transformação de aceleração elimina mensagens com base no momento em que as processa, encaminhando no máximo uma mensagem por padrão de tópico dentro de cada intervalo configurado. Use a limitação para proteger os sistemas a jusante de fontes de ráfagas ou de alta frequência sem alterar o conteúdo da mensagem.

Para uma visão geral dos gráficos de fluxo de dados e de como as transformações se compõem num pipeline, consulte Visão Geral dos Gráficos de Fluxo de Dados.

Pré-requisitos

Uma instância do Operações IoT do Azure implementada num cluster Kubernetes. Para mais informações, consulte Deploy Operações IoT do Azure.

A implementação cria automaticamente um endpoint de registo predefinido chamado default que aponta para mcr.microsoft.com. As transformações incorporadas usam este endpoint.

Limitação de escalamento para grafos com estado

Importante

As transformadas de janela e de acelerador são com estado. Cada instância mantém o seu próprio estado e as instâncias não partilham esse estado entre si. Quando o número de instâncias do perfil de fluxo de dados é superior a um, as subscrições partilhadas distribuem mensagens entre instâncias, de modo que cada instância vê apenas um subconjunto das mensagens. Uma transformada de janela calcula então agregações como médias, somas e contagens sobre um conjunto de dados parcial, e uma transformada de aceleração impõe o limite de taxa configurado de forma independente em cada instância, em vez de em todo o pipeline.

Defina a contagem de instâncias do perfil de fluxo de dados para 1 para qualquer grafo de fluxo de dados que utilize uma transformação de janela ou de aceleração. Grafos de fluxo de dados sem estado que utilizam apenas transformadas de mapeamento, filtro, ramificação e concatenação podem usar com segurança contagens de instâncias mais elevadas para aumentar o rendimento.

Como funciona a transformação do acelerador

A transformada de aceleração avalia o tema de cada mensagem recebida contra uma lista ordenada de regras por tópico:

O primeiro jogo vence. A transformação avalia as regras por ordem. A primeira regra cujo topic padrão corresponde ao tema da mensagem determina se a transformação encaminha a mensagem. A transformação não verifica regras posteriores, mesmo que também coincidam.
Os tópicos sem correspondência passam adiante. Se nenhuma regra corresponder ao tema da mensagem, a transformação encaminha a mensagem sem qualquer limitação.
O encaminhamento é feito com base no tempo, não na contagem. Para uma regra coincidente, a transformação encaminha a primeira mensagem que processa e, em seguida, elimina todas as mensagens subsequentes com esse padrão até que decorra o intervalo configurado (1000 / maxMessagesPerSecond milissegundos, arredondado para cima) desde a última mensagem encaminhada. Este comportamento limita a taxa máxima, mas não permite que os bursts compensem quedas anteriores.
Estado partilhado por padrão, não por tópico. Quando o padrão de topic uma regra usa um coringa, todos os tópicos concretos que lhe correspondem partilham o mesmo estado de aceleração dentro de uma instância de transformação. Por exemplo, uma única sensors/+ regra limita a taxa combinada entre sensors/temperature e sensors/humidity, não cada uma independentemente.
0 deixa cair tudo. Definir maxMessagesPerSecond para 0 uma regra elimina todas as mensagens que correspondem ao padrão dessa regra.
O tempo baseia-se no tempo de processamento, não no conteúdo da mensagem. A transformação utiliza o seu relógio monotónico próprio quando processa cada mensagem. Não lê um campo de marca temporal da carga útil da mensagem.

Note

A transformação do acelerador decide apenas se encaminha ou deixa uma mensagem em falta. Nunca modifica o conteúdo da mensagem.

Configurar regras de aceleração por tópico

Defina regras de aceleração na throttle chave de configuração (não rules) como um objeto JSON com um perTopicThrottles array.

Na configuração de transformação de acelerador, adicione uma ou mais regras de aceleração. Para cada regra, especifique:

Setting	Description
Topic	O padrão de tópicos do MQTT para corresponder. Suporta os carateres universais `+` (de nível único) e `#` (multinível).
Máximo de mensagens por segundo	A taxa máxima de reencaminhamento para tópicos que correspondem a este padrão. Defina para `0` eliminar todas as mensagens correspondentes.

A CLI aplica todo o grafo a partir de um ficheiro de configuração. Adicione esta configuração aos nós configuration da transformação de aceleração no seu graph.json ficheiro e aplique usando az iot ops dataflowgraph apply.

A configuração é um objeto JSON:

{
  "perTopicThrottles": [
    {
      "topic": "sensors/temperature",
      "maxMessagesPerSecond": 10
    }
  ]
}

Adicione esta configuração no campo value como uma cadeia escapada, sob a chave throttle:

"configuration": [
  {
    "key": "throttle",
    "value": "{\"perTopicThrottles\":[{\"topic\":\"sensors/temperature\",\"maxMessagesPerSecond\":10}]}"
  }
]

Sugestão

Para gerar a string escapada, guarde as regras num ficheiro como rules.json, execute jq -c . rules.json, e cole a saída de linha única no value campo.

configuration: [
  {
    key: 'throttle'
    value: '{"perTopicThrottles":[{"topic":"sensors/temperature","maxMessagesPerSecond":10}]}'
  }
]

Importante

A utilização dos manifestos de implementação do Kubernetes não é suportada em ambientes de produção e deve ser usada apenas para depuração e testes.

configuration:
  - key: throttle
    value: |
      {
        "perTopicThrottles": [
          {
            "topic": "sensors/temperature",
            "maxMessagesPerSecond": 10
          }
        ]
      }

Com esta regra, a transformação encaminha a primeira sensors/temperature mensagem que processa, depois elimina quaisquer mensagens adicionais sobre esse tema que processe menos de 100 milissegundos depois (1000 / 10). Esta regra não afeta outros temas além sensors/temperaturede .

Cada entrada em perTopicThrottles tem as seguintes propriedades:

Property	Required	Description
`topic`	Sim	Padrão de tópicos MQTT para corresponder. Suporta os `+` wildcards (de nível único) e `#` (multinível, apenas de retrocesso).
`maxMessagesPerSecond`	Sim	Taxa máxima de encaminhamento para tópicos que correspondem a este padrão, em mensagens por segundo. A transformada permite valores fracionários. Por exemplo, `0.1` permite uma mensagem a cada 10 segundos. Deve ser zero ou um número positivo e finito. Defina para `0` eliminar todas as mensagens que correspondam ao padrão. A temporização tem precisão de 1 milissegundo, por isso valores superiores a `1000` têm o mesmo limite efetivo que `1000`.

Importante

Cada padrão topic em perTopicThrottles deve ser único. A transformação rejeita configurar a mesma cadeia de padrão mais do que uma vez durante a inicialização.

Use múltiplas regras por tema

Como a transformação avalia as regras por ordem e o primeiro jogo vence, liste padrões mais específicos antes dos mais gerais se quiseres que tenham o seu próprio limite de taxa:

Adicione duas regras, nesta ordem:

Order	Topic	Máximo de mensagens por segundo
1	`sensors/temperature`	`10`
2	`sensors/#`	`1`

{
  "perTopicThrottles": [
    {
      "topic": "sensors/temperature",
      "maxMessagesPerSecond": 10
    },
    {
      "topic": "sensors/#",
      "maxMessagesPerSecond": 1
    }
  ]
}

configuration: [
  {
    key: 'throttle'
    value: '{"perTopicThrottles":[{"topic":"sensors/temperature","maxMessagesPerSecond":10},{"topic":"sensors/#","maxMessagesPerSecond":1}]}'
  }
]

Importante

A utilização dos manifestos de implementação do Kubernetes não é suportada em ambientes de produção e deve ser usada apenas para depuração e testes.

configuration:
  - key: throttle
    value: |
      {
        "perTopicThrottles": [
          { "topic": "sensors/temperature", "maxMessagesPerSecond": 10 },
          { "topic": "sensors/#", "maxMessagesPerSecond": 1 }
        ]
      }

As mensagens em sensors/temperature correspondem à primeira regra e estão limitadas a 10 mensagens por segundo. Mensagens sobre qualquer outro sensors/* tema (por exemplo, sensors/humidity) correspondem à segunda regra e partilham um limite combinado de 1 mensagem por segundo.

Importante

A ordem é relevante. Se a sensors/# regra fosse listada primeiro, também corresponderia sensors/temperature às mensagens, e a transformação nunca chegaria à regra mais específica.

Use coringas para limitar grupos de tópicos

O topic padrão suporta os mesmos curingas que os filtros de tópicos do MQTT:

carácter universal	Resultados	Example
`+`	Exatamente um nível de tópico	`sensors/+/status` corresponde a `sensors/line1/status` mas não a `sensors/line1/sub/status`
`#`	Zero ou mais níveis temáticos restantes, e deve ser o último segmento	`sensors/#` corresponde a `sensors`, `sensors/temperature` e `sensors/line1/temperature`

Todos os tópicos concretos que correspondem à mesma regra do wildcard partilham um mesmo estado de aceleração.

Adicione uma regra com o tópico sensors/+ e o máximo de mensagens por segundo 1.

{
  "perTopicThrottles": [
    {
      "topic": "sensors/+",
      "maxMessagesPerSecond": 1
    }
  ]
}

configuration: [
  {
    key: 'throttle'
    value: '{"perTopicThrottles":[{"topic":"sensors/+","maxMessagesPerSecond":1}]}'
  }
]

Importante

A utilização dos manifestos de implementação do Kubernetes não é suportada em ambientes de produção e deve ser usada apenas para depuração e testes.

configuration:
  - key: throttle
    value: |
      {
        "perTopicThrottles": [
          { "topic": "sensors/+", "maxMessagesPerSecond": 1 }
        ]
      }

Com esta regra, uma mensagem no sensors/temperature e uma mensagem no sensors/humidity processadas com 500 milissegundos de intervalo não são ambas reencaminhadas — a segunda é descartada, porque conta para o mesmo intervalo partilhado de 1 segundo que a primeira, independentemente do tópico específico em que esteja.

Note

Um padrão nu # corresponde a todos os tópicos e aplica um único limite de taxa combinado em cada instância de transformação.

Eliminar todas as mensagens de um tópico

Defina maxMessagesPerSecond para 0 eliminar todas as mensagens que correspondam a um padrão, sem remover a regra ou o tópico do seu pipeline:

Adicione uma regra com o tópico debug/# e o máximo de mensagens por segundo 0.

{
  "perTopicThrottles": [
    {
      "topic": "debug/#",
      "maxMessagesPerSecond": 0
    }
  ]
}

configuration: [
  {
    key: 'throttle'
    value: '{"perTopicThrottles":[{"topic":"debug/#","maxMessagesPerSecond":0}]}'
  }
]

Importante

A utilização dos manifestos de implementação do Kubernetes não é suportada em ambientes de produção e deve ser usada apenas para depuração e testes.

configuration:
  - key: throttle
    value: |
      {
        "perTopicThrottles": [
          { "topic": "debug/#", "maxMessagesPerSecond": 0 }
        ]
      }

Implemente um gráfico de fluxo de dados com potência

Para aplicar a limitação de ponta a ponta, implemente um grafo de fluxo de dados que ligue uma fonte, uma transformação de aceleração e um destino. Use a ferramenta que se adapte ao seu fluxo de trabalho.

Na experiência de Operações, crie-se um gráfico de fluxo de dados com uma transformada de aceleração:

Adicione uma fonte que leia do seu tópico MQTT.
Adiciona uma transformação de acelerador . Adicione uma ou mais regras por tema, ordenadas do mais ao menos específico.
Adiciona um destino que envie para o tópico de saída.

A CLI do Azure aplica um grafo de fluxo de dados a partir de um único ficheiro de configuração JSON. Cria um graph.json ficheiro com as propriedades do grafo. No graph.json ficheiro, cada transformação armazena a sua configuração no value campo como uma string JSON escapada. Para a forma legível da configuração do acelerador, veja Configurar regras de aceleração por tópico. Aplica-o usando az iot ops dataflowgraph apply.

{
  "mode": "Enabled",
  "nodes": [
    {
      "nodeType": "Source",
      "name": "sensors",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "sensors/#"
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "throttle",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-throttle:1.0.0",
        "configuration": [
          {
            "key": "throttle",
            "value": "{\"perTopicThrottles\":[{\"topic\":\"sensors/temperature\",\"maxMessagesPerSecond\":10},{\"topic\":\"sensors/#\",\"maxMessagesPerSecond\":1}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Destination",
      "name": "output",
      "destinationSettings": {
        "endpointRef": "default",
        "dataDestination": "telemetry/throttled"
      }
    }
  ],
  "nodeConnections": [
    {
      "from": {
        "name": "sensors"
      },
      "to": {
        "name": "throttle"
      }
    },
    {
      "from": {
        "name": "throttle"
      },
      "to": {
        "name": "output"
      }
    }
  ]
}

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  name: 'sensors-throttle'
  parent: dataflowProfile
  properties: {
    mode: 'Enabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensors'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [ 'sensors/#' ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'throttle'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-throttle:1.0.0'
          configuration: [
            {
              key: 'throttle'
              value: '{"perTopicThrottles":[{"topic":"sensors/temperature","maxMessagesPerSecond":10},{"topic":"sensors/#","maxMessagesPerSecond":1}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Destination'
        name: 'output'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'telemetry/throttled'
        }
      }
    ]
    nodeConnections: [
      { from: { name: 'sensors' }, to: { name: 'throttle' } }
      { from: { name: 'throttle' }, to: { name: 'output' } }
    ]
  }
}

Importante

A utilização dos manifestos de implementação do Kubernetes não é suportada em ambientes de produção e deve ser usada apenas para depuração e testes.

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: sensors-throttle
  namespace: azure-iot-operations
spec:
  profileRef: default
  nodes:
    - nodeType: Source
      name: sensors
      sourceSettings:
        endpointRef: default
        dataSources:
          - sensors/#
    - nodeType: Graph
      name: throttle
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-throttle:1.0.0
        configuration:
          - key: throttle
            value: |
              {
                "perTopicThrottles": [
                  { "topic": "sensors/temperature", "maxMessagesPerSecond": 10 },
                  { "topic": "sensors/#", "maxMessagesPerSecond": 1 }
                ]
              }
    - nodeType: Destination
      name: output
      destinationSettings:
        endpointRef: default
        dataDestination: telemetry/throttled
  nodeConnections:
    - from: { name: sensors }
      to: { name: throttle }
    - from: { name: throttle }
      to: { name: output }

Limitações

Não modifica as mensagens. A transformação do acelerador apenas encaminha ou corta mensagens; Não altera o conteúdo da mensagem.
O primeiro jogo vence. A transformação aplica-se apenas à primeira regra cujo topic padrão corresponde ao tema da mensagem. Lista padrões mais específicos antes dos mais gerais.
Estado partilhado por padrão. Uma regra wildcard partilha o seu limite de taxa em todos os temas concretos que corresponde. Não há um limite separado por tema.
Sem autorização para explosões. A transformação impõe um tempo mínimo entre mensagens encaminhadas por padrão correspondente. Não acumula capacidade não utilizada de períodos anteriores e mais lentos.
Precisão de milissegundos. O intervalo mínimo de aceleração é de 1 milissegundo, pelo que valores superiores a 1000 para maxMessagesPerSecond não aumentam a taxa efetiva de encaminhamento para além de 1.000 mensagens por segundo.
O tempo baseia-se no tempo de processamento, não no conteúdo da mensagem. A transformação utiliza o tempo em que processa cada mensagem, não a hora em que o corretor a recebeu ou um campo de carimbo temporal na carga útil.
O estado é local e é armazenado em memória. Cada instância do perfil de fluxo de dados mantém o seu próprio estado de aceleração. Reiniciar ou reconfigurar a transformação reinicia esse estado. Se um perfil tiver múltiplas instâncias, cada instância aplica a taxa configurada de forma independente.
Não são permitidos padrões de tópicos duplicados. Configurar a mesma topic cadeia de caracteres mais do que uma vez em perTopicThrottles falha quando a transformação é inicializada.

Comentários

Esta página foi útil?

Last updated on 2026-08-14

Limitar dados em gráficos de fluxo de dados

Pré-requisitos

Limitação de escalamento para grafos com estado

Como funciona a transformação do acelerador

Configurar regras de aceleração por tópico

Use múltiplas regras por tema

Use coringas para limitar grupos de tópicos

Eliminar todas as mensagens de um tópico

Implemente um gráfico de fluxo de dados com potência

Limitações

Conteúdo relacionado

Comentários

Recursos adicionais