[redis] Postgresql(timescaledb)에 저장된 데이터를 시간단위로 redis에 옮기기

안녕하세요 오랜만에 개발관련글을 올립니다.

 

이번 포스팅할 내용은 redis입니다. 

 

레디스에 대해서는 오래전부터 알고 있었지만 쓸 일이 없기도 하고 굳이 해봐야하나 싶은 생각이 있어 건드리지 않았었습니다. 만

 

뭐 이것 저것 건드리다보니 사용하게 되었는데요

 

오늘 할 예제는 timescaledb에 저장되어있는 최신 시계열 데이터를 redis에 옮기는 작업입니다.

 

사실 이게 맞는 구조는 아닌거같습니다. 왜냐하면 redis 자체가 인메모리 디비로써 그 목적성이 캐싱에 있다고하는데(물론 요새는 램성능이 워낙좋아져서 redis 자체를 통합DB로 사용하는 경우도 있다고합니다.) 실시간으로 저장되는 timescaldb에서 다시 최신데이터를 빼워서 redis에 저장한다?? 이건좀.. 비효율적인거를 떠나서 이렇게 구조를 짤 바에는 안하는게 나은가 싶은데요. 

 

*물론 실시간성이 중요하지 않다면 그래프나 표를 보여주는 시각화쪽에서는 좀 이득이 있을거 같긴합니다.

*만약 실시간성이 정말 중요하다면 timescaledb자체도 좋긴하지만 애초에 저장하는 모듈에서 2가지로 저장할 거 같습니다. 축적용 DB와 많은 조회가 일어나는 DB용으로 말이죠

 

일단은 아예 쓸 모 없지는 않을 거 같고, 연습 겸 구현을 해봤습니다.

 

대략적인 구조는 다음과 같습니다.

 

 

 

애초에 레디스를 사용하는 이유는 데이터를 보여주는 웹에서 많은 양의 데이터를 timescaledb에서 조회하다보니 너무 느려서 웹에 표시되는 그래프나 차트가 느렸기 때문입니다.

 

 

이번 구현은 chat gpt를 사용해서 한 번 해봤는데요. 저는 개인적으로 정말 잘 쓰고 있고 이번에 결제까지 했습니다.

 

 

레디스 구축부터 해봅시다.

 

참고로 chat gpt에 물어보면 이렇게 알려줍니다.

 

 

저는 뭐 도커로 구축할거라 이거는 그냥 무시했구요. (사실 궁금해서 쳐봄) chat gpt데이터는 21년 9월까지 데이터라고 하니 여러분들도 너무 무지성으로 뭐해달라 뭐해달라 하진 마세요.

 

아무튼 redis 구축을 위해 다음 명령어를 서버에 쳐줍시다.

 

docker run --name redis_test -d -p 6379:6379 redis:latest --requirepass "비밀번호"

 

requirepass 옵션을 통해 레디스 비밀번호를 설정해주면 좋겠죠??

 

그 다음 python 코드를 부탁해봅시다.

 

 

만족스러운 코드를 작성해줬지만 중복을 피하고싶은데 그 방법이 없군요? 다시 물어봅시다.

 

 

 

 

이 외에도 여러가지 개념적인 것도 물어봤습니다.

존재하는 키에 대해서 중복저장이 되는지와

 

특정 시간이 지나면 사라지게 할 수 있는지 등

 

 

 

그래서 최종적으로 gpt가 알려준 코드와 그것을 수정한 제 코드는 다음과 같습니다.

 

 

 

 

import psycopg2
import redis
from datetime import datetime, timedelta
import time
import pprint
from psycopg2.extras import RealDictCursor
import json

# TimescaleDB 연결 정보
db_host = ""
db_port = ""
db_name = ""
db_user = ""
db_password = ""

# Redis 연결 정보
redis_host = ""
redis_port = 
redis_password = "!"
redis_db = 0

# PostgreSQL 연결 설정


timescaledb_connection = psycopg2.connect(
    host=db_host,
    port=db_port,
    dbname=db_name,
    user=db_user,
    password=db_password,
)

# Redis 연결 설정
redis_connection = redis.StrictRedis(
    host=redis_host, port=redis_port, db=0, password=redis_password
)


# 마지막 저장된 데이터의 시간을 Redis에서 가져오기
def get_last_stored_time():
    last_stored_time = redis_connection.get("last_stored_time")
    if last_stored_time:
        return last_stored_time.decode()
    return None


# 해당 시간 이후의 데이터를 TimescaleDB에서 조회하기
def fetch_data_since(time):
    query = """SELECT * FROM table WHERE "time" > %s;"""
    with timescaledb_connection.cursor(cursor_factory=RealDictCursor) as cursor:
        cursor.execute(query, (time,))
        data = cursor.fetchall()
        # pprint.pprint(data)
        # exit()

    return data


# 조회된 데이터를 Redis에 저장하기
def store_data_in_redis(key, data):
    expiration_time = 300
    serialized_data = json.dumps(data)
    redis_connection.set(key, serialized_data, ex=expiration_time)


# 메인 작업
def main():
    last_stored_time = get_last_stored_time()

    # 마지막 저장된 데이터의 시간이 없을 경우 현재 시간을 사용합니다.
    if not last_stored_time:
        last_stored_time = datetime.now() - timedelta(seconds=5)

    data_since_last_stored_time = fetch_data_since(last_stored_time)

    # Redis에 데이터 저장
    for row in data_since_last_stored_time:

        row["time"] = row["time"].strftime("%Y-%m-%d %H:%M:%S")
        timestamp = row["time"]

        key = f"data_{timestamp}"
        store_data_in_redis(key, row)

        # Redis에 마지막으로 저장된 데이터의 시간 업데이트
        print("last_stored_time : ", timestamp)
        redis_connection.set("last_stored_time", timestamp, ex=10)



if __name__ == "__main__":
    while True:
        main()
        print("작동중")
        time.sleep(4)

 

저장할 데이터가 dict형태라 json 형태로 변환도 해주고 datetime 형태는 저장이안되서 str형식으로 바꿔주기도 하는 등 여러가지를 또 chat gpt에 물어보면서 해결했습니다. 서칭하는데 속도가 말도안되게 빨라진 기분입니다.

 

요거를 그대로 실행해보면 

 

아 참고로 저는 redis 모니터링을 위해 redis desktop manager라는 툴을 사용했습니다.

 

 

 

 

왼쪽에 보면 데이터가 저장되고있고 미리 설정해놓은 시간에 맞춰서 지워지고있습니다.

 

 

 

네 이렇게 오늘은 redis를 구축해보고 여기에 키값을 저장하는 것 까지 해봤습니다. 용어자체가 헷갈리긴 하지만 역시 db는 거기서 거기인거 같네요.

목적에 따라서 적절한 db를 사용할 수 있도록 공부해야겠습니다. 그럼 수고하십쇼