Computer >> 컴퓨터 >  >> 프로그래밍 >> Redis

Redis Streams를 활용한 시계열 데이터 처리 완벽 가이드

레디스(Redis)는 흔히 소수의 간단한 데이터 구조 프리미티브에 네트워크 인터페이스를 제공하는 "데이터 구조 서버"로 인식되어 왔습니다. 그런데 스트림(Streams)은 레디스가 오래전 정렬 집합(Sorted Set)을 도입한 이후 처음으로 등장한 주요 범용 데이터 구조입니다. 이번 글에서는 이 새로운 구조의 핵심 활용 분야 중 하나인 시계열(time series) 데이터 모델링 방법을 알아보겠습니다.

스트림(Streams): 레디스의 새로운 데이터 구조

Redis 스트림은 키-값 쌍으로 구성된 추가 전용(append-only) 시계열입니다.

클라이언트는 동시에 몇 개든 하나의 스트림에 데이터를 기록할 수 있으며, 쓰기 작업이 발생할 때마다 시계열에 삽입된 항목에 대한 고유하고 항상 오름차순인 ID를 받게 됩니다.

데이터를 읽는 클라이언트는 새 데이터가 들어올 때까지 대기(블로킹)하거나, 마지막으로 읽은 메시지에 대한 "북마크"를 유지하며 배치 처리를 할 수 있고, 더 복잡한 "소비자 그룹(consumer group)"으로 구성해 작업 부하를 분산하고 메시지 확인(acknowledgment)까지 수행할 수도 있습니다.

스트림은 다룰 내용이 방대하기 때문에 이 글에서는 시계열 데이터 모델링에 활용하는 방법만 간략히 살펴봅니다. 새로운 데이터 구조 덕분에 과거 List나 Sorted Set 타입으로 시계열을 모델링하던 방식보다 훨씬 간단해졌습니다.

여기서 사용할 코드 예제에서는 node.js용 ioredis 클라이언트를 사용합니다:

var Redis = require('ioredis');
var redis = new Redis();

스트림에 데이터 보내기

스트림은 추가 전용이므로 일반적으로 필요한 정보는 데이터를 기록할 Redis 키 이름과 키-값 쌍 집합뿐입니다.

여기서는 공기 질 센서의 측정값을 기록해 보겠습니다. site:pdx 키로 키-값 쌍 스트림을 전송하며, 구체적으로 현재 대기 질 지수(AQI)와 섭씨 온도를 보냅니다:

redis.xadd('site:pdx', '*',
           'aqi', 37,
           'tempc', 5.1).then(function(id) {
  console.log("id:", id);
});

> 1527974818120-0

새로운 측정값을 기록할 때마다 XADD 명령을 보냅니다. 응답으로 받는 것은 쿼리에 활용할 수 있는 고유하면서 항상 오름차순인 ID입니다. ID의 첫 번째 부분인 1527974818120은 Redis 서버가 부여한 타임스탬프이고, 두 번째 부분은 여러 클라이언트가 동시에 쓰기를 시도할 때 충돌을 피하기 위한 증가 숫자입니다.

위 예제처럼 쓰기 명령의 두 번째 인수로 *를 지정하면 Redis가 자체적으로 타임스탬프를 지정해 데이터를 저장하도록 합니다.

데이터를 쓸 때 직접 타임스탬프를 지정할 수도 있지만 일반적으로 권장되지 않습니다. Redis가 타임스탬프를 선택하게 하면 많은 클라이언트가 ID 선택이나 순서 조율 없이 단일 스트림에 동시에 데이터를 기록할 수 있습니다. Redis 서버가 이런 세부 사항을 클라이언트 대신 처리해 주기 때문입니다.

간단한 읽기

스트림에 값이 어느 정도 쌓이면 ID나 타임스탬프 범위를 지정해 값 범위를 조회할 수 있습니다. 이는 내 공기 질 지표의 최근 측정값을 그래프로 표시하는 앱 등에 활용할 수 있습니다:

redis.xrange('site:pdx',
             '1527974818120-0',
             '+',
             'COUNT', 5).then(function(resp) {

  // resp now holds 5 readings, pass them to the open graph:
  // console.log(resp);
});

> [ [ '1543947167906-0', [ 'aqi', '31', 'tempc', '5.1' ] ],
> [ '1543947168312-0', [ 'aqi', '31', 'tempc', '5.3' ] ],
> [ '1543947168901-0', [ 'aqi', '31', 'tempc', '5.4' ] ],
> [ '1543947170033-0', [ 'aqi', '31', 'tempc', '5.4' ] ],
> [ '1543947171460-0', [ 'aqi', '31', 'tempc', '5.6' ] ] ]

스트림의 어느 위치에서든 숫자 범위를 샘플링할 수 있으므로, 그래프 시스템이 성능 저하 없이 과거 데이터를 조회할 수 있습니다.

블로킹과 폴링

데이터 범위 조회는 그래프나 이력 모니터링에 유용하지만, 때로는 들어오는 데이터에 즉시 반응하는 시스템을 만들고 싶을 때도 있습니다. 이럴 때에도 Redis 스트림은 XREAD를 통해 탁월하게 부합합니다:

redis.xread('BLOCK', 10000,
            'STREAMS', 'site:pdx', '$').then(function(resp) {

  // close the windows if aqi > 50
  console.log(resp);
});

이런 블로킹 연산은 데이터가 들어오거나 타임아웃(위 예제에서는 10,000ms)이 될 때까지 대기합니다. 따라서 지속적인 폴링을 유지하려면 위 예제처럼 데이터가 준비될 때까지 블로킹 상태로 기다렸다가, 데이터를 받거나 명령이 타임아웃될 때마다 동일한 XREAD 명령을 다시 호출하면 됩니다.

스트림 연결 사이에 데이터 누락을 방지하려면, 읽을 때 마지막으로 읽은 ID를 함께 넘겨주면 됩니다. 그러면 정확히 마지막 지점부터 이어서 데이터를 가져올 수 있습니다.

위 예제에서는 스트림에 들어오는 어떤 데이터든 상관없었기 때문에 "새 데이터만"을 의미하는 특수 토큰 '$'를 사용했습니다. 특정 ID를 지정하면 명령이 아주 약간 달라질 뿐입니다:

redis.xread('BLOCK', 10000,
            'STREAMS', 'site:pdx', '1543947171460-0');

여러 개의 서로 다른 스트림을 한 번에 읽을 수도 있으며, 이 경우 먼저 데이터가 도착한 스트림의 값을 반환합니다:

redis.xread('BLOCK', 10000,
            'STREAMS',
              'site:pdx', 'site:global',
              '1543947171460-0', '$');

이 예제에서 명령은 site:pdx에 ID 1543947171460-0보다 최신 데이터가 기록되거나, site:global에 새로운 데이터가 기록되는 즉시 반환됩니다.

소비자 조율

스트림을 통해 많은 데이터가 흐르기 시작하면, 인바운드 메시지를 처리하기 위해 여러 개의 소비자 복제본을 운용하고 싶어질 수 있습니다. 이런 소비자들은 메시지를 가져와 작업을 수행한 후 작업이 완료되었음을 "확인(acknowledge)"해야 합니다. Redis 스트림은 이러한 연산을 위한 프리미티브도 제공합니다. 세부 내용이 많아 여기서는 다루지 않으며, 자세한 사항은 공식 문서를 참고하시기 바랍니다.

스트림 명령 빠른 참고

스트림을 위해 새로 추가된 Redis 명령은 매우 많습니다! 아래는 빠른 참고 목록이며, 자세한 내용은 공식 문서에서 확인할 수 있습니다.

기본 명령:

  • XADD: 스트림에 항목(키-값 쌍 묶음) 추가
  • XRANGE 및 XREVRANGE: 범위 선택 또는 스트림 항목 순회
  • XREAD: 특정 ID보다 최신 항목 조회(선택적으로 블로킹)
  • XTRIM: 오래된 항목을 버려 스트림 크기 절감
  • XDEL: 스트림에서 특정 항목 삭제
  • XLEN: 스트림의 항목 수 계산
  • XINFO: 스트림 메타데이터 검사

소비자 그룹 명령:

  • XGROUP: 소비자 그룹 생성, 삭제, 초기화 및 멤버 제거
  • XREADGROUP: XREAD와 유사하지만 소비자 그룹을 통해 메시지 수신
  • XPENDING: 소비자 그룹에 전달되었지만 ack되지 않은 메시지 검사
  • XACK: 소비자 그룹의 메시지를 확인 처리하여 대기(pending) 목록에서 제거
  • XCLAIM: 종료된(dead) 소비자에게서 메시지 인수

활용 사례

스트림의 가장 많이 언급되는 활용 사례는 IoT 워크로드입니다. 센서가 데이터를 스트림에 올리고, 소비자들이 이를 분석, 콜드 스토리지 아카이빙, 그래프 표시 등 다양한 용도로 소비하는 구조입니다. 이 경우 크기가 제한된(capped) 스트림이 특히 유용합니다. 고정 크기의 스트림을 할당해 예측 가능한 메모리 사용량을 확보할 수 있기 때문입니다.

스트림은 기존에 다른 Redis 데이터 구조를 사용하던 애플리케이션에도 적합합니다. Celery나 Sidekiq 같은 큐잉 애플리케이션은 스트림의 소비자 그룹을 활용해 Redis 네이티브 방식으로 읽기 영수증(read receipt)을 검사할 수 있습니다. 또한 Redis pub/sub으로 구현된 간단한 채팅 앱을 소개하는 블로그 글이 많은데, pub/sub은 메시지가 클라이언트에 게시된 후에는 보관되지 않으므로 Redis Streams를 사용하면 훨씬 더 견고한 구조로 만들 수 있습니다.

직접 사용해 볼 준비가 되셨나요?

새로운 RedisGreen 서버에서 몇 번의 클릭만으로 Redis 스트림을 바로 사용해 볼 수 있습니다.