time_window_rolling_avg_fl()

이 함수 time_window_rolling_avg_fl() 는 일정 기간 동안 필요한 값의 롤링 평균을 계산하는 UDF(사용자 정의 함수) 입니다.

정규 시계열에 대한 상수 기간 동안의 롤링 평균 계산(즉, 상수 간격 포함)은 상수 시간 창을 동일한 계수의 고정 너비 필터로 변환할 수 있으므로 series_fir()를 사용하여 수행할 수 있습니다. 그러나 창의 실제 샘플 수가 다르기 때문에 불규칙한 시계열에 대해 계산하는 것이 더 복잡합니다. 여전히 강력한 검사 연산자를 사용하여 달성할 수 있습니다.

이러한 유형의 롤링 창 계산은 메트릭 값이 변경될 때만 내보내지는 사용 사례(일정 간격이 아님)에 필요합니다. 예를 들어 에지 디바이스가 변경 시 클라우드로 메트릭을 전송하여 통신 대역폭을 최적화하는 IoT의 경우입니다.

구문

T | invoke time_window_rolling_avg_fl(, t_col y_col, key_col, dt [, 방향 ])

구문 규칙에 대해 자세히 알아봅니다.

매개 변수

이름	Type	필수	설명
t_col	string	✔️	레코드의 타임스탬프를 포함하는 열의 이름입니다.
y_col	string	✔️	레코드의 메트릭 값을 포함하는 열의 이름입니다.
key_col	string	✔️	레코드의 파티션 키를 포함하는 열의 이름입니다.
dt	timespan	✔️	롤링 창의 기간입니다.
direction	int		집계 방향입니다. 가능한 값은 +1 또는 -1입니다. 롤링 창은 각각 현재 시간 앞/뒤로 설정됩니다. 기본값은 -1이며, 이전 버전 롤링 창은 스트리밍 시나리오에 사용할 수 있는 유일한 방법입니다.

함수 정의

다음과 같이 해당 코드를 쿼리 정의 함수로 포함하거나 데이터베이스에 저장된 함수로 만들어 함수를 정의할 수 있습니다.

쿼리 정의

다음 let 문을 사용하여 함수를 정의합니다. 사용 권한이 필요 없습니다.

Important

let 문은 자체적으로 실행할 수 없습니다. 그 뒤에 테이블 형식 식 문이 있어야 합니다. 작업 예제 time_window_rolling_avg_fl()를 실행하려면 예제를 참조 하세요.

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.

저장

다음을 사용하여 저장 함수를 한 번 정의합니다 .create function. 데이터베이스 사용자 권한이 필요합니다.

Important

예제와 같이 함수를 사용하려면 먼저 이 코드를 실행하여 함수를 만들어야 합니다.

.create-or-alter function with (folder = "Packages\\Series", docstring = "Time based rolling average of a metric")
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 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)

저장

Important

이 예제를 성공적으로 실행하려면 먼저 함수 정의 코드를 실행하여 함수를 저장해야 합니다.

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)

출력

timestamp	value	avg_value	key
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

8:05의 첫 번째 값(10)은 10분 후방 창에서 떨어진 단일 값만 포함하며, 두 번째 값(15)은 8:09 및 8:05에 두 샘플의 평균입니다.