Window는 무한한 스트림 데이터를 시간 또는 이벤트 단위로 구간화하여 처리하기 위한 개념이다.
스트림 데이터는 끝이 없기 때문에 특정 기간 동안의 데이터를 묶어 집계하거나 조인하기 위해 윈도우가 필요하다.
Rowtime
윈도우 연산에서 시간 기준은 매우 중요하며 ksqlDB는 다음 시간을 RowTime으로 사용할 수 있다.
- Event Time: Producer가 전송한 시간이다.
- Ingestion Time: Kafka에 레코드가 적재된 시간이다.
- Processing Time: 쿼리가 실행되는 시점의 시스템 시간이다.
Broker의 log.message.timestamp.type 또는 topic의 message.timestamp.type 속성값은 CreateTime(Producer 메시지 전송 시각), LogAppendTime(Broker Topic 저장 시각)으로 이들을 설정할 수 있음
ksqlDB는 기본적으로 Kafka Record의 timestamp를 ROWTIME으로 사용하며 필요 시 컬럼을 Rowtime으로 지정하여 Custom Rowtime을 구성할 수 있다.
Window
윈도우는 스트림 데이터를 시간 기반으로 분리하고, 각 구간에 대해 독립적인 집계를 수행한다.
윈도우 연산은 WINDOWSTART, WINDOWEND 컬럼을 통해 구간을 표현한다.
Tumbling Window
고정된 크기의 시간 구간으로 나뉘며, 윈도우 간 중복이 없다.
SELECT order_zipcode,
FROM_UNIXTIME(WINDOWSTART) AS window_start,
FROM_UNIXTIME(WINDOWEND) AS window_end,
COUNT(orderId) AS order_cnt
FROM orders
WINDOW TUMBLING (SIZE 5 MINUTE, GRACE PERIOD 1 MINUTE)
GROUP BY order_zipcode
EMIT CHANGES;Hopping Window
윈도우 크기는 고정이지만 Advance 간격으로 이동하여 윈도우가 서로 겹친다.
SELECT regionId,
COUNT(*) AS cnt
FROM pageviews
WINDOW HOPPING (SIZE 30 SECONDS, ADVANCE BY 10 SECONDS)
GROUP BY regionId
EMIT CHANGES;Session Window
이벤트 사이의 비활동 시간(inactivity gap)을 기준으로 동적으로 윈도우가 형성된다.
SELECT regionId,
COUNT(*) AS cnt
FROM pageviews
WINDOW SESSION (60 SECONDS)
GROUP BY regionId
EMIT CHANGES;Grace Period
Grace Period는 지연 도착(late arrival) 데이터를 허용하는 기간이다.
윈도우 종료 후에도 Grace 기간 안에 들어오면 동일 윈도우에 포함된다.
지정하지 않으면 기본 Grace Period는 24시간이다.
Window Table
-
Window가 적용된 테이블의 key 컬럼은 window값을 반영하여 생성되며 json/avro format 형으로 지정 권장
-
Window절은 반드시 Group by와 함께 사용해야 함. 만약 window레벨로 group by를 하고자 한다면 Dummy 값을 Group by 절에 사용하여 적용 가능.
-
Window가 적용된 테이블에 다시 Permanent Query(CTAS)를 수행할 수 없음
Window JOIN
윈도우 JOIN은 두 스트림 또는 스트림-테이블 간 결합 시 시간 범위를 고려하여 매칭하는 방식이다.
특히 Stream-Stream 조인에서는 반드시 WITHIN 절을 지정해야 한다.
SELECT o.order_id,
o.total_amount,
s.shipment_id
FROM new_orders o
JOIN shipments s
WITHIN 1 HOURS
ON o.id = s.order_id
EMIT CHANGES;윈도우 조인에서는 RowTime이 매칭 조건의 기준이 된다.
시간 범위를 벗어난 레코드는 조인되지 않는다.
JOIN & Rowtime
| 조인 유형 | 조인 결과 Rowtime |
|---|---|
| Stream-Stream | 두 데이터 중 가장 최근 Rowtime을 가진다 |
| Stream-Table | 두 데이터 중 Stream의 Rowtime을 가진다 |
| Table-Table | 두 데이터 중 가장 최근 Rowtime을 가진다 |
CREATE STREAM device_status_monitor_stream AS
SELECT ...
FROM device_master_stream b
JOIN device_status_stream a
WITHIN 5 MINUTES
GRACE PERIOD 1 MINUTE
ON a.device_id = b.device_id
EMIT CHANGES;이 조인에서 최종 Rowtime은
- Stream-Stream 조인의 규칙을 따라
a와b중 더 최근 Rowtime을 결과 Rowtime으로 사용한다.
