timeSeriesLastTwoSamples

PromQL 스타일의 irate와 idelta 계산을 위해 시계열 데이터를 재샘플링하는 집계 함수예요. 타임스탬프와 값의 쌍으로 시계열 데이터를 받아 최근 샘플 2개만 저장해요.

출처: 문서

본문

timeSeriesLastTwoSamples는 PromQL 스타일의 irate와 idelta 계산을 위해 시계열 데이터를 재샘플링하는 집계 함수예요. v25.6.0에 도입되었어요.

타임스탬프와 값의 쌍으로 시계열 데이터를 받아 최근 샘플 2개까지만 저장하는 집계 함수예요. 이 집계 함수는 Materialized View와, 그리드에 정렬된 타임스탬프로 재샘플링된 시계열 데이터를 저장하는 Aggregated 테이블과 함께 사용하도록 설계되었어요.

집계 테이블은 각 정렬된 타임스탬프에 대해 마지막 2개 값만 저장해요. 덕분에 원시 테이블에 저장된 데이터보다 훨씬 적은 데이터를 읽어 PromQL 스타일의 irate와 idelta를 계산할 수 있어요.

이 함수는 비공개 프리뷰(private preview) 상태라 enable_time_series_aggregate_functions=true 로 설정해야 해요.

구문 (Syntax)

timeSeriesLastTwoSamples(timestamp, value)

인자 (Arguments)

timestamp — 샘플의 타임스탬프예요. DateTime 또는 DateTime64 또는 (U)Int* 또는 Int* 타입이에요.
value — 타임스탬프에 대응하는 시계열 값이에요. Float32 또는 Float64 타입이에요.

반환 값 (Returned value)

길이가 0부터 2까지인 두 배열의 쌍을 반환해요. 첫 번째 배열은 샘플링된 시계열의 타임스탬프를, 두 번째 배열은 대응하는 시계열 값을 담아요.

Tuple(Array(DateTime), Array(Float64)) 타입이에요.

예시 (Examples)

원시 데이터용 예시 테이블과 재샘플링 데이터 저장용 테이블

쿼리:

SET enable_time_series_aggregate_functions = 1;
-- 원시 데이터용 테이블
DROP TABLE IF EXISTS t_raw_timeseries;

CREATE TABLE t_raw_timeseries
(
    metric_id UInt64,
    timestamp DateTime64(3, 'UTC') CODEC(DoubleDelta, ZSTD),
    value Float64 CODEC(DoubleDelta)
)
ENGINE = MergeTree()
ORDER BY (metric_id, timestamp);

-- 더 큰(15초) 시간 간격으로 재샘플링한 데이터 테이블
DROP TABLE IF EXISTS t_resampled_timeseries_15_sec;

CREATE TABLE t_resampled_timeseries_15_sec
(
    metric_id UInt64,
    grid_timestamp DateTime('UTC') CODEC(DoubleDelta, ZSTD), -- 15초에 정렬된 타임스탬프
    samples AggregateFunction(timeSeriesLastTwoSamples, DateTime64(3, 'UTC'), Float64)
)
ENGINE = AggregatingMergeTree()
ORDER BY (metric_id, grid_timestamp);

-- 재샘플링 테이블을 채우는 MV
DROP VIEW IF EXISTS mv_resampled_timeseries;

CREATE MATERIALIZED VIEW mv_resampled_timeseries TO t_resampled_timeseries_15_sec
(
    metric_id UInt64,
    grid_timestamp DateTime('UTC') CODEC(DoubleDelta, ZSTD),
    samples AggregateFunction(timeSeriesLastTwoSamples, DateTime64(3, 'UTC'), Float64)
)
AS SELECT
    metric_id,
    ceil(toUnixTimestamp(timestamp + interval 999 millisecond) / 15, 0) * 15 AS grid_timestamp, -- 타임스탬프를 다음 그리드 지점으로 올림
    initializeAggregation('timeSeriesLastTwoSamplesState', timestamp, value) AS samples
FROM t_raw_timeseries
ORDER BY metric_id, grid_timestamp;

-- 데이터 일부 삽입
INSERT INTO t_raw_timeseries(metric_id, timestamp, value) SELECT number%10 AS metric_id, '2024-12-12 12:00:00'::DateTime64(3, 'UTC') + interval ((number/10)%100)*900 millisecond as timestamp, number%3+number%29 AS value FROM numbers(1000);

-- 원시 데이터 확인
SELECT *
FROM t_raw_timeseries
WHERE metric_id = 3 AND timestamp BETWEEN '2024-12-12 12:00:12' AND '2024-12-12 12:00:31'
ORDER BY metric_id, timestamp;

응답:

3	2024-12-12 12:00:12.870	29
3	2024-12-12 12:00:13.770	8
3	2024-12-12 12:00:14.670	19
3	2024-12-12 12:00:15.570	30
3	2024-12-12 12:00:16.470	9
3	2024-12-12 12:00:17.370	20
3	2024-12-12 12:00:18.270	2
3	2024-12-12 12:00:19.170	10
3	2024-12-12 12:00:20.070	21
3	2024-12-12 12:00:20.970	3
3	2024-12-12 12:00:21.870	11
3	2024-12-12 12:00:22.770	22
3	2024-12-12 12:00:23.670	4
3	2024-12-12 12:00:24.570	12
3	2024-12-12 12:00:25.470	23
3	2024-12-12 12:00:26.370	5
3	2024-12-12 12:00:27.270	13
3	2024-12-12 12:00:28.170	24
3	2024-12-12 12:00:29.069	6
3	2024-12-12 12:00:29.969	14
3	2024-12-12 12:00:30.869	25

타임스탬프 '2024-12-12 12:00:15'와 '2024-12-12 12:00:30'의 마지막 2개 샘플 조회하기

쿼리:

SET enable_time_series_aggregate_functions = 1;
-- 원시 데이터용 테이블
DROP TABLE IF EXISTS t_raw_timeseries;

CREATE TABLE t_raw_timeseries
(
    metric_id UInt64,
    timestamp DateTime64(3, 'UTC') CODEC(DoubleDelta, ZSTD),
    value Float64 CODEC(DoubleDelta)
)
ENGINE = MergeTree()
ORDER BY (metric_id, timestamp);

-- 더 큰(15초) 시간 간격으로 재샘플링한 데이터 테이블
DROP TABLE IF EXISTS t_resampled_timeseries_15_sec;

CREATE TABLE t_resampled_timeseries_15_sec
(
    metric_id UInt64,
    grid_timestamp DateTime('UTC') CODEC(DoubleDelta, ZSTD), -- 15초에 정렬된 타임스탬프
    samples AggregateFunction(timeSeriesLastTwoSamples, DateTime64(3, 'UTC'), Float64)
)
ENGINE = AggregatingMergeTree()
ORDER BY (metric_id, grid_timestamp);

-- 재샘플링 테이블을 채우는 MV
DROP VIEW IF EXISTS mv_resampled_timeseries;

CREATE MATERIALIZED VIEW mv_resampled_timeseries TO t_resampled_timeseries_15_sec
(
    metric_id UInt64,
    grid_timestamp DateTime('UTC') CODEC(DoubleDelta, ZSTD),
    samples AggregateFunction(timeSeriesLastTwoSamples, DateTime64(3, 'UTC'), Float64)
)
AS SELECT
    metric_id,
    ceil(toUnixTimestamp(timestamp + interval 999 millisecond) / 15, 0) * 15 AS grid_timestamp, -- 타임스탬프를 다음 그리드 지점으로 올림
    initializeAggregation('timeSeriesLastTwoSamplesState', timestamp, value) AS samples
FROM t_raw_timeseries
ORDER BY metric_id, grid_timestamp;

-- 데이터 일부 삽입
INSERT INTO t_raw_timeseries(metric_id, timestamp, value) SELECT number%10 AS metric_id, '2024-12-12 12:00:00'::DateTime64(3, 'UTC') + interval ((number/10)%100)*900 millisecond as timestamp, number%3+number%29 AS value FROM numbers(1000);

-- 재샘플링 데이터 확인
SELECT metric_id, grid_timestamp, (finalizeAggregation(samples).1 as timestamp, finalizeAggregation(samples).2 as value)
FROM t_resampled_timeseries_15_sec
WHERE metric_id = 3 AND grid_timestamp BETWEEN '2024-12-12 12:00:15' AND '2024-12-12 12:00:30'
ORDER BY metric_id, grid_timestamp;

응답:

3	2024-12-12 12:00:15	(['2024-12-12 12:00:14.670','2024-12-12 12:00:13.770'],[19,8])
3	2024-12-12 12:00:30	(['2024-12-12 12:00:29.969','2024-12-12 12:00:29.069'],[14,6])

원시 데이터에서 idelta와 irate 계산하기

쿼리:

SET enable_time_series_aggregate_functions = 1;
-- 원시 데이터용 테이블
DROP TABLE IF EXISTS t_raw_timeseries;

CREATE TABLE t_raw_timeseries
(
    metric_id UInt64,
    timestamp DateTime64(3, 'UTC') CODEC(DoubleDelta, ZSTD),
    value Float64 CODEC(DoubleDelta)
)
ENGINE = MergeTree()
ORDER BY (metric_id, timestamp);

-- 데이터 일부 삽입
INSERT INTO t_raw_timeseries(metric_id, timestamp, value) SELECT number%10 AS metric_id, '2024-12-12 12:00:00'::DateTime64(3, 'UTC') + interval ((number/10)%100)*900 millisecond as timestamp, number%3+number%29 AS value FROM numbers(1000);

-- 집계 테이블은 15초로 정렬된 각 타임스탬프에 대해 마지막 2개 값만 저장합니다.
-- 덕분에 원시 테이블에 저장된 것보다 훨씬 적은 데이터를 읽어 PromQL 스타일의 irate와 idelta를 계산할 수 있습니다.

WITH
    '2024-12-12 12:00:15'::DateTime64(3,'UTC') AS start_ts,       -- 타임스탬프 그리드의 시작
    start_ts + INTERVAL 60 SECOND AS end_ts,   -- 타임스탬프 그리드의 끝
    15 AS step_seconds,   -- 타임스탬프 그리드의 스텝
    45 AS window_seconds  -- "staleness" 윈도우
SELECT
    metric_id,
    timeSeriesInstantDeltaToGrid(start_ts, end_ts, step_seconds, window_seconds)(timestamp, value),
    timeSeriesInstantRateToGrid(start_ts, end_ts, step_seconds, window_seconds)(timestamp, value)
FROM t_raw_timeseries
WHERE metric_id = 3 AND timestamp BETWEEN start_ts - interval window_seconds seconds AND end_ts
GROUP BY metric_id;

응답:

3	[11,8,-18,8,11]	[12.222222222222221,8.88888888888889,1.1111111111111112,8.88888888888889,12.222222222222221]

재샘플링된 데이터에서 idelta와 irate 계산하기

쿼리:

SET enable_time_series_aggregate_functions = 1;
-- 원시 데이터용 테이블
DROP TABLE IF EXISTS t_raw_timeseries;

CREATE TABLE t_raw_timeseries
(
    metric_id UInt64,
    timestamp DateTime64(3, 'UTC') CODEC(DoubleDelta, ZSTD),
    value Float64 CODEC(DoubleDelta)
)
ENGINE = MergeTree()
ORDER BY (metric_id, timestamp);

-- 더 큰(15초) 시간 간격으로 재샘플링한 데이터 테이블
DROP TABLE IF EXISTS t_resampled_timeseries_15_sec;

CREATE TABLE t_resampled_timeseries_15_sec
(
    metric_id UInt64,
    grid_timestamp DateTime('UTC') CODEC(DoubleDelta, ZSTD), -- 15초에 정렬된 타임스탬프
    samples AggregateFunction(timeSeriesLastTwoSamples, DateTime64(3, 'UTC'), Float64)
)
ENGINE = AggregatingMergeTree()
ORDER BY (metric_id, grid_timestamp);

-- 재샘플링 테이블을 채우는 MV
DROP VIEW IF EXISTS mv_resampled_timeseries;

CREATE MATERIALIZED VIEW mv_resampled_timeseries TO t_resampled_timeseries_15_sec
(
    metric_id UInt64,
    grid_timestamp DateTime('UTC') CODEC(DoubleDelta, ZSTD),
    samples AggregateFunction(timeSeriesLastTwoSamples, DateTime64(3, 'UTC'), Float64)
)
AS SELECT
    metric_id,
    ceil(toUnixTimestamp(timestamp + interval 999 millisecond) / 15, 0) * 15 AS grid_timestamp, -- 타임스탬프를 다음 그리드 지점으로 올림
    initializeAggregation('timeSeriesLastTwoSamplesState', timestamp, value) AS samples
FROM t_raw_timeseries
ORDER BY metric_id, grid_timestamp;

-- 데이터 일부 삽입
INSERT INTO t_raw_timeseries(metric_id, timestamp, value) SELECT number%10 AS metric_id, '2024-12-12 12:00:00'::DateTime64(3, 'UTC') + interval ((number/10)%100)*900 millisecond as timestamp, number%3+number%29 AS value FROM numbers(1000);

WITH
    '2024-12-12 12:00:15'::DateTime64(3,'UTC') AS start_ts,       -- 타임스탬프 그리드의 시작
    start_ts + INTERVAL 60 SECOND AS end_ts,   -- 타임스탬프 그리드의 끝
    15 AS step_seconds,   -- 타임스탬프 그리드의 스텝
    45 AS window_seconds  -- "staleness" 윈도우
SELECT
    metric_id,
    timeSeriesInstantDeltaToGrid(start_ts, end_ts, step_seconds, window_seconds)(timestamps, values),
    timeSeriesInstantRateToGrid(start_ts, end_ts, step_seconds, window_seconds)(timestamps, values)
FROM (
    SELECT
        metric_id,
        finalizeAggregation(samples).1 AS timestamps,
        finalizeAggregation(samples).2 AS values
    FROM t_resampled_timeseries_15_sec
    WHERE metric_id = 3 AND grid_timestamp BETWEEN start_ts - interval window_seconds seconds AND end_ts
)
GROUP BY metric_id;

응답:

3	[11,8,-18,8,11]	[12.222222222222221,8.88888888888889,1.1111111111111112,8.88888888888889,12.222222222222221]

timeSeriesLastTwoSamples

timeSeriesLastTwoSamples

본문

더 알아보기 (Learn more)