Partilhar via

Enriqueça dados usando fluxos de dados

Sugestão

Os grafos de fluxo de dados suportam enriquecimento com capacidades expandidas, incluindo enriquecimento em transformações de filtro e ramificação. Para novos projetos que utilizam endpoints MQTT, Kafka ou OpenTelemetry, consulte Adicionar dados externos em grafos de fluxo de dados.

Você pode enriquecer dados usando a função de conjuntos de dados de contextualização. Quando os registros de entrada são processados, você pode consultar esses conjuntos de dados com base nas condições relacionadas aos campos do registro de entrada. Esta capacidade permite interações dinâmicas. Os dados desses conjuntos de dados podem ser usados para complementar informações nos campos de saída e participar de cálculos complexos durante o processo de mapeamento.

Para carregar dados de exemplo no repositório de estado, use o CLI do repositório de estado.

Por exemplo, considere o seguinte conjunto de dados com alguns registros, usando o formato de linhas JSON:

{ "Position": "Analyst", "BaseSalary": 70000, "WorkingHours": "Regular" }
{ "Position": "Receptionist", "BaseSalary": 43000, "WorkingHours": "Regular" }

O mapeador acessa o conjunto de dados de referência armazenado no repositório de estado de Operações do Azure IoT usando um valor de chave com base em uma condição especificada na configuração de mapeamento. Os nomes das chaves no armazenamento de estado correspondem a um conjunto de dados na configuração do fluxo de dados.

datasets: [
  {
    key: 'position',
    inputs: [
      '$source.Position' //  - $1
      '$context.Position' // - $2
    ],
    expression: '$1 == $2'
  }
]

Quando um novo registro está sendo processado, o mapeador executa as seguintes etapas:

  • Solicitação de dados: o mapeador envia uma solicitação ao armazenamento de estado para recuperar o conjunto de dados armazenado sob a chave Position.
  • Correspondência de registros: em seguida, o mapeador consulta esse conjunto de dados para encontrar o primeiro registro em que o Position campo no conjunto de dados corresponde ao Position campo do registro de entrada.
{
  inputs: [
    '$context(position).WorkingHours' //  - $1 
  ]
  output: 'WorkingHours'
}
{
  inputs: [
    'BaseSalary' // - - - - - - - - - - - - $1
    '$context(position).BaseSalary' //  - - $2
  ]
  output: 'BaseSalary'
  expression: 'if($1 == (), $2, $1)'
}

Neste exemplo, o WorkingHours campo é adicionado ao registro de saída, enquanto o BaseSalary é usado condicionalmente somente quando o registro de entrada não contém o BaseSalary campo (ou o valor é null se for um campo anulável). A solicitação dos dados de contextualização não acontece com todos os registros recebidos. O mapeador solicita o conjunto de dados e, em seguida, recebe notificações do armazenamento de estado sobre as alterações, enquanto usa uma versão em cache do conjunto de dados.

É possível usar vários conjuntos de dados:

datasets: [
  {
    key: 'position'
    inputs: [
      '$source.Position'  // - $1
      '$context.Position' // - $2
    ],
    expression: '$1 == $2'
  }
  {
    key: 'permissions'
    inputs: [
      '$source.Position'  // - $1
      '$context.Position' // - $2
    ],
    expression: '$1 == $2'
  }
]

Em seguida, use as referências misturadas:

inputs: [
  '$context(position).WorkingHours'  // - $1
  '$context(permissions).NightShift' // - $2
]

As referências de entrada usam a chave do conjunto de dados como position ou permission. Se a chave no armazenamento de estado for inconveniente de usar, você pode definir um alias:

datasets: [
  {
    key: 'datasets.parag10.rule42 as position'
    inputs: [
      '$source.Position'  // - $1
      '$context.Position' // - $2
    ],
    expression: '$1 == $2'
  }
]

A configuração renomeia o conjunto de dados com a chave datasets.parag10.rule42 para position.

  • Last updated on