Usar os últimos valores conhecidos em um pipeline do processador de dados

Use o último estágio de valor conhecido (LKV) em um pipeline do processador de dados para manter um registro atualizado e completo de seus dados. O estágio LKV rastreia os valores mais recentes de pares chave-valor para mensagens no pipeline. O estágio pode então enriquecer mensagens usando os valores LKV rastreados. O último rastreamento de valor conhecido e o enriquecimento são importantes para processos a jusante que dependem de:

• Vários pontos de dados de séries cronológicas em um carimbo de data/hora específico.
• Cargas úteis que sempre têm um valor para uma chave específica.

Em um pipeline de processamento de dados, o estágio LKV é um estágio opcional. Quando você usa o estágio LKV, você pode:

• Adicione vários estágios LKV a um pipeline. Cada estágio LKV pode rastrear vários valores.
• Enriqueça as mensagens com os valores LKV armazenados, garantindo que os dados permaneçam completos e abrangentes.
• Mantenha os LKVs atualizados automaticamente com os valores mais recentes das mensagens recebidas.
• Rastreie LKVs separadamente para cada partição lógica. O estágio LKV opera independentemente em cada partição lógica.
• Configure o tempo de expiração para cada LKV rastreado para gerenciar a duração para que ele permaneça válido. Esse controle ajuda a garantir que as mensagens não sejam enriquecidas com valores obsoletos.

O estágio LKV mantém a integridade cronológica dos dados. O estágio garante que mensagens com carimbos de data/hora anteriores não substituam ou substituam LKVs por mensagens com carimbos de data/hora posteriores.

O estágio LKV enriquece as mensagens recebidas com os últimos valores conhecidos que rastreia. Esses valores enriquecidos representam dados registrados anteriormente e não são necessariamente os valores atuais em tempo real. Certifique-se de que esse comportamento esteja alinhado com suas expectativas de processamento de dados.

Pré-requisitos

Para configurar e usar um estágio de pipeline agregado, você precisa de uma instância implantada do processador de dados que inclua o componente opcional do processador de dados.

Configurar o palco

A configuração JSON do estágio LKV define os detalhes do estágio. Para criar o palco, você pode interagir com a interface do usuário baseada em formulário ou fornecer a configuração JSON na guia Avançado :

Se você incluir o caminho do carimbo de data/hora, isso o ajudará a entender com precisão quando os LKVs foram registrados e aumentará a transparência e a rastreabilidade.

inputPath igual a outputPath

A mensagem de saída é o valor real da mensagem ou LKV se a chave rastreada estiver ausente da carga útil da mensagem. Qualquer valor de entrada tem prioridade e o estágio não o substitui por um LKV. Para identificar se o valor da mensagem é um valor LKV, use o caminho do carimbo de data/hora. O caminho do carimbo de data/hora só será incluído na mensagem de saída se o valor na mensagem for o LKV rastreado.

inputPath não é igual a outputPath

O palco grava o LKV para todas outputPath as mensagens recebidas. Use essa configuração para controlar a diferença entre os valores nas cargas úteis de mensagens subsequentes.

Configuração de exemplo

O exemplo a seguir mostra uma mensagem de exemplo para o estágio LKV com a mensagem chegando em 10:02 e com uma carga útil que contém o valor LKV rastreado .payload.temperature :

{ 
  { 
    "systemProperties":{ 
        "partitionKey":"pump", 
        "partitionId":5, 
        "timestamp":"2023-01-11T10:02:07Z" 
    }, 
    "qos":1, 
    "topic":"/assets/pump/#" 
    }, 
    "payload":{ 
        "humidity": 10, 
        "temperature":250, 
        "pressure":30, 
        "runningState": true 
    } 
} 

Configuração do LKV:

Os valores de LKV rastreados são:

• .payload.temperature é 250.
• O carimbo de data/hora do LKV é 2023-01-11T10:02:07Z

Para uma mensagem que chega às 11:05 com uma carga útil que não tem a propriedade temperatura, o estágio LKV enriquece a mensagem com os valores rastreados:

Exemplo de entrada para o estágio LKV às 11:05:

{ 
    "systemProperties":{ 
        "partitionKey":"pump", 
        "partitionId":5, 
        "timestamp":"2023-01-11T11:05:00Z" 
    }, 
    "qos":1, 
    "topic":"/assets/pump/#" 
    }, 
    "payload":{ 
        "runningState": true 
    } 
} 

Exemplo de saída do estágio LKV às 11:05:

{ 
    "systemProperties":{ 
        "partitionKey":"pump", 
        "partitionId":5, 
        "timestamp":"2023-01-11T11:05:00Z" 
    }, 
    "qos":1, 
    "topic":"/assets/pump/#" 
    }, 
    "payload":{ 
        "lkvtemperature":250, 
        "lkvtemperature_timestamp"":"2023-01-11T10:02:07Z" 
        "runningState": true 
    } 
} 

Conteúdos relacionados

• Agregar dados em um pipeline
• Enriqueça dados em um pipeline
• Filtrar dados em um pipeline
• Chamar para um ponto de extremidade gRPC a partir de um pipeline
• Chamar para um ponto de extremidade HTTP a partir de um pipeline
• Transformar dados em um pipeline