Configurar um estágio de origem MQ em um pipeline do processador de dados

O estágio de origem é o primeiro e necessário estágio em um pipeline de processador de dados. O estágio de origem recebe dados no pipeline de processamento de dados e os prepara para processamento posterior. O estágio de origem MQ permite que você assine mensagens de um tópico MQTT. No estágio de origem, você define detalhes de conexão com a fonte MQ e estabelece uma configuração de particionamento com base em seus requisitos específicos de processamento de dados.

Pré-requisitos

• Uma instância implantada do processador de dados que inclui o componente opcional do processador de dados.
• Uma instância do broker MQTT com todos os dados brutos necessários disponíveis é operacional e acessível.

Configurar o código-fonte MQ

Para configurar a origem MQ:

• Forneça detalhes de conexão para a fonte MQ. Essa configuração inclui o tipo da fonte MQ, a URL do broker MQTT, o nível de Qualidade de Serviço (QoS), o tipo de sessão e os tópicos a serem assinados.
• Especifique o método de autenticação. Atualmente limitado a autenticação baseada em nome de usuário/senha ou token de conta de serviço.

A tabela a seguir descreve os parâmetros de configuração de origem MQ:

Para saber mais sobre segredos, consulte Gerenciar segredos para sua implantação do Azure IoT Operations Preview.

O processador de dados não reordena dados fora de ordem provenientes do broker MQTT. Se os dados forem recebidos fora de ordem do corretor, eles permanecerão assim no pipeline.

Selecionar formato de dados

Em um pipeline do processador de dados, o campo de formato no estágio de origem especifica como desserializar os dados de entrada. Por padrão, o pipeline do processador de dados usa o raw formato que significa que ele não converte os dados de entrada. Para usar muitos recursos do processador de dados, como Filter estágios ou Enrich estágios em um pipeline, você deve desserializar seus dados no estágio de entrada. Você pode optar por desserializar seus dados de entrada de , , , CSV, ou Protobuf formatos em uma mensagem legível do processador de dados para usar a funcionalidade completa do processador de JSONdados. CBORMessagePackjsonStream

As tabelas a seguir descrevem as diferentes opções de configuração de desserialização:

O Data Format campo é obrigatório e o seu valor determina os outros campos obrigatórios.

Para desserializar mensagens CSV, você também precisa especificar os seguintes campos:

Para desserializar mensagens Protobuf, você também precisa especificar os seguintes campos:

Configurar particionamento

O particionamento em um pipeline divide os dados de entrada em partições separadas. O particionamento permite o paralelismo de dados no pipeline, o que pode melhorar a taxa de transferência e reduzir a latência. As estratégias de particionamento afetam a forma como os dados são processados nos outros estágios do pipeline. Por exemplo, o último estágio de valor conhecido e o estágio agregado operam em cada partição lógica.

Para particionar seus dados, especifique uma estratégia de particionamento e o número de partições a serem usadas:

O processador de dados adiciona metadados adicionais à mensagem de entrada. Consulte Visão geral da estrutura de mensagens do processador de dados para entender como especificar corretamente a expressão de particionamento que é executada na mensagem de entrada. Por padrão, a expressão de particionamento é definida como 0 com o tipo Partition para ID enviar todos os dados de entrada para uma única partição.

Para obter recomendações e saber mais, consulte O que é particionamento?.

Configuração de exemplo

A seguir mostra um exemplo de configuração para o estágio:

Em seguida, essa configuração gera mensagens parecidas com o exemplo a seguir:

{
    "Timestamp": "2023-08-10T00:54:58.6572007Z", 
    "MessageType": "ua-deltaframe",
    "payload": {
      "temperature": {
        "SourceTimestamp": "2023-08-10T00:54:58.2543129Z",
        "Value": 7109
      },
      "Tag 10": {
        "SourceTimestamp": "2023-08-10T00:54:58.2543482Z",
        "Value": 7109
      }
    },
    "DataSetWriterName": "oven",
    "SequenceNumber": 4660
}

Conteúdos relacionados

• Formatos de serialização e desserialização
• O que é particionamento?