Partilhar via

time_window_rolling_avg_fl()

  • Artigo

Aplica-se a: ✅Microsoft Fabric✅Azure Data Explorer✅Azure Monitor✅Microsoft Sentinel

A função time_window_rolling_avg_fl() é uma UDF (função definida pelo usuário) que calcula a média móvel do valor necessário em uma janela de tempo de duração constante.

O cálculo da média móvel em uma janela de tempo constante para séries temporais regulares (ou seja, com intervalos constantes) pode ser obtido usando series_fir(), pois a janela de tempo constante pode ser convertida em um filtro de largura fixa de coeficientes iguais. No entanto, calculá-lo para séries temporais irregulares é mais complexo, pois o número real de amostras na janela varia. Ainda assim, isso pode ser alcançado usando o poderoso operador de varredura .

Esse tipo de cálculo de janela contínua é necessário para casos de uso em que os valores de métrica são emitidos somente quando alterados (e não em intervalos constantes). Por exemplo, na IoT, onde os dispositivos de borda enviam métricas para a nuvem apenas após alterações, otimizando a largura de banda de comunicação.

Sintaxe

T | invoke time_window_rolling_avg_fl(, t_col y_col, key_col, dt [, direção ])

Saiba mais sobre as convenções de sintaxe.

Parâmetros

Nome Digitar Obrigatória Descrição
t_col string ✔️ O nome da coluna que contém o carimbo de data/hora dos registros.
y_col string ✔️ O nome da coluna que contém o valor da métrica dos registros.
key_col string ✔️ O nome da coluna que contém a chave de partição dos registros.
dt timespan ✔️ A duração da janela de interrupção.
direction int A direção de agregação. Os valores possíveis são +1 ou -1. Uma janela contínua é definida a partir do tempo atual para frente/para trás, respectivamente. O padrão é -1, pois a janela de rolagem regressiva é o único método possível para cenários de streaming.

Definição de função

Você pode definir a função inserindo seu código como uma função definida por consulta ou criando-a como uma função armazenada em seu banco de dados, da seguinte maneira:

Defina a função usando a instrução let a seguir. Nenhuma permissão é necessária.

Importante

Uma instrução let não pode ser executada sozinha. Ele deve ser seguido por uma instrução de expressão tabular. Para executar um exemplo funcional de time_window_rolling_avg_fl(), consulte Exemplo.

let time_window_rolling_avg_fl=(tbl:(*), t_col:string, y_col:string, key_col:string, dt:timespan, direction:int=int(-1))
{
    let tbl_ex = tbl | extend timestamp = column_ifexists(t_col, datetime(null)), value = column_ifexists(y_col, 0.0), key = column_ifexists(key_col, '');
    tbl_ex 
    | partition hint.strategy=shuffle by key 
    (
        extend timestamp=pack_array(timestamp, timestamp - direction*dt), delta = pack_array(-direction, direction)
        | mv-expand timestamp to typeof(datetime), delta to typeof(long)
        | sort by timestamp asc, delta desc    
        | scan declare (cum_sum:double=0.0, cum_count:long=0) with 
        (
            step s: true => cum_count = s.cum_count + delta, 
                            cum_sum = s.cum_sum + delta * value; 
        )
        | extend avg_value = iff(direction == 1, prev(cum_sum)/prev(cum_count), cum_sum/cum_count)
        | where delta == -direction 
        | project timestamp, value, avg_value, key
    )
};
// Write your query to use the function here.

Exemplo

O exemplo a seguir usa o operador invoke para executar a função.

Para usar uma função definida por consulta, invoque-a após a definição da função inserida.

let time_window_rolling_avg_fl=(tbl:(*), t_col:string, y_col:string, key_col:string, dt:timespan, direction:int=int(-1))
{
    let tbl_ex = tbl | extend timestamp = column_ifexists(t_col, datetime(null)), value = column_ifexists(y_col, 0.0), key = column_ifexists(key_col, '');
    tbl_ex 
    | partition hint.strategy=shuffle by key 
    (
        extend timestamp=pack_array(timestamp, timestamp - direction*dt), delta = pack_array(-direction, direction)
        | mv-expand timestamp to typeof(datetime), delta to typeof(long)
        | sort by timestamp asc, delta desc    
        | scan declare (cum_sum:double=0.0, cum_count:long=0) with 
        (
            step s: true => cum_count = s.cum_count + delta, 
                            cum_sum = s.cum_sum + delta * value; 
        )
        | extend avg_value = iff(direction == 1, prev(cum_sum)/prev(cum_count), cum_sum/cum_count)
        | where delta == -direction 
        | project timestamp, value, avg_value, key
    )
};
let tbl = datatable(ts:datetime,  val:real, key:string) [
    datetime(8:00), 1, 'Device1',
    datetime(8:01), 2, 'Device1',
    datetime(8:05), 3, 'Device1',
    datetime(8:05), 10, 'Device2',
    datetime(8:09), 20, 'Device2',
    datetime(8:40), 4, 'Device1',
    datetime(9:00), 5, 'Device1',
    datetime(9:01), 6, 'Device1',
    datetime(9:05), 30, 'Device2',
    datetime(9:50), 7, 'Device1'
];
tbl
| invoke time_window_rolling_avg_fl('ts', 'val', 'key', 10m)

Saída

timestamp value avg_value chave
2021-11-29 08:05:00.0000000 10 10 Device2
2021-11-29 08:09:00.0000000 20 15 Device2
2021-11-29 09:05:00.0000000 30 30 Device2
2021-11-29 08:00:00.0000000 1 1 Device1
2021-11-29 08:01:00.0000000 2 1.5 Device1
2021-11-29 08:05:00.0000000 3 2 Device1
2021-11-29 08:40:00.0000000 4 4 Device1
2021-11-29 09:00:00.0000000 5 5 Device1
2021-11-29 09:01:00.0000000 6 5.5 Device1
2021-11-29 09:50:00.0000000 7 7 Device1

O primeiro valor (10) às 8:05 contém apenas um único valor, que caiu na janela de 10 minutos para trás, o segundo valor (15) é a média de duas amostras às 8:09 e às 8:05, etc.