timesacle db + fastapi

https://www.youtube.com/watch?v=Se5ipte9DMY&t=3s

 

timesacleDB

timesacle DB 는 PostgreSQL 을 기반의 TimescaleDB는 오픈소스 기반의 시계열(time-series) 데이터베이스로 시계열 데이터를 효율적으로 저장, 관리, 분석할 수 있는 기능을 제공하며, PostgreSQL의 모든 기능을 그대로 사용할 수 있는 확장(extension) 형태로 동작한다.

 

 

주요 특징

1. PostgreSQL의 확장이므로, PostgreSQL의 모든 기능과 호환 (SQL 쿼리, 트랜잭션, 인덱스, 사용자 정의 함수 등을 그대로 활용)
2. 시간 축을 기준으로 데이터를 효율적으로 분할 및 저장 (데이터의 쓰기/읽기 성능을 최적화하기 위해 자동 파티셔닝과 압축을 제공)
3. Hypertable
   1. TimescaleDB의 핵심 개념으로, 시계열 데이터를 관리하기 위한 논리적 테이블
   2. 실제로는 여러 개의 파티션 테이블로 분리되어 있지만, 사용자에게는 하나의 테이블처럼 보임
4. 내장 분석 도구
   1. 시계열 데이터에 특화된 여러 SQL 함수(time_bucket, locf, percentile_cont 등)를 제공
   2. 데이터 집계, 다운샘플링(down-sampling), 예측 등을 쉽게 구현

단점

• 완전히 독립적인 데이터베이스가 아니며 PostgreSQL에 의존.
• 극단적인 초대규모 분산 아키텍처에서는 다른 솔루션(예: ClickHouse, InfluxDB)이 더 적합할 수 있음.

---

1) 시계열 데이터

• 시계열 데이터는 "시간을 통해 순차적으로 발생한 관측치의 집합" 의미
• 관측치 데이터는 서로 "연결성, 관련" 이있고, 더 정확하게는 "고정된 시간 구간의 관측치" 여야 유의미한 시계열 데이터 이며, 시차(Time lag)가 동일해야 한다. (불규칙한 시간 구간을 유의미한 시계열 데이터로 볼 순 없다.)

 

2) 기존 시계열 데이터 처리 DBMS

https://db-engines.com/en/ranking_trend/time+series+dbms

 

 

3) 하이퍼 테이블

하이퍼테이블은 시간별로 데이터를 자동으로 분할하는 데이터를 자동으로 분할합니다. 하이퍼테이블은 일반 포스트그레SQL 테이블과 동일한 방식으로 상호 작용합니다. 일반 PostgreSQL 테이블처럼 동작하지만, 내부적으로는 여러 개의 물리적 테이블(청크)로 분할되어 저장됩니다.

청크(Chunk)

• Chunk는 Hypertable의 물리적 저장 단위
• 데이터 범위별로 분리되어 있어, 읽기/쓰기 및 쿼리 성능이 향상
• TimescaleDB는 필요한 Chunk만 읽어오는 최적화된 쿼리 플래너를 제공

Chunk 크기 조정

• 예상되는 데이터 삽입 속도와 쿼리 패턴을 고려하여 설정.
• 너무 작으면 청크 관리 오버헤드 증가, 너무 크면 쿼리 성능 저하.

-- Hypertable 생성 시 Chunk 크기를 1개월로 설정
SELECT create_hypertable('sensor_data', 'time', chunk_time_interval => INTERVAL '1 month');

오래된 청크 압축

-- Hypertable 압축 활성화
ALTER TABLE sensor_data SET (timescaledb.compress, true);

-- 특정 기간 이전의 데이터 압축
SELECT compress_chunk(show_chunks('sensor_data', older_than => INTERVAL '6 months'));

 

데이터 수명 관리

-- 오래된 청크 삭제
SELECT drop_chunks('sensor_data', older_than => INTERVAL '1 year');

 

 

번외 TimescaleDB와 Apache Druid 비교

1. 기본 개념

• TimescaleDB:
  • PostgreSQL 기반의 확장(extension)으로, SQL 친화적인 시계열 데이터베이스
  • 트랜잭션 및 관계형 데이터베이스 기능을 제공하며, SQL을 그대로 사용할 수 있다.
  • OLTP(Online Transaction Processing)와 OLAP(Online Analytical Processing)을 혼합한 사용에 적합
• Apache Druid:
  • 분산형 데이터 스토리지 및 분석 플랫폼으로, 대규모 시계열 데이터의 고속 쿼리에 특화
  • 이벤트 기반 데이터 저장 및 OLAP(주로 분석)에 최적화

---

 

 

 

fastapi와 timescaledb 실습

docker-compose.yml 파일을 만들어 TimescaleDB를 설정합니다. (postgres위에 동작하여 대부분의 파라미터가 동일하다)

version: "3.8"

services:
  timescaledb:
    image: timescale/timescaledb:latest-pg15
    container_name: timescaledb
    ports:
      - "5432:5432"
    environment:
      POSTGRES_USER: fastapi_user
      POSTGRES_PASSWORD: fastapi_pass
      POSTGRES_DB: fastapi_db

디렉토리 구조

project/
├── main.py
├── models.py
└── database.py

database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "postgresql://fastapi_user:fastapi_pass@localhost:5432/fastapi_db"

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# Dependency for database session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

models.py

from sqlalchemy import Column, Integer, String, Float, TIMESTAMP
from sqlalchemy.sql import func
from database import Base

class SensorData(Base):
    __tablename__ = "sensor_data"

    id = Column(Integer, primary_key=True, index=True)
    time = Column(TIMESTAMP, server_default=func.now(), index=True)
    sensor_name = Column(String, index=True)
    value = Column(Float)

main.py

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from database import Base, engine, get_db
from models import SensorData
from pydantic import BaseModel

app = FastAPI()

# Create tables
Base.metadata.create_all(bind=engine)

@app.get("/")
def index():
    return "hello"

# Pydantic 모델 정의
class SensorDataRequest(BaseModel):
    sensor_name: str
    value: float

@app.post("/sensor-data/")
def create_sensor_data(data: SensorDataRequest, db: Session = Depends(get_db)):
    sensor_data = SensorData(sensor_name=data.sensor_name, value=data.value)
    db.add(sensor_data)
    db.commit()
    db.refresh(sensor_data)
    return sensor_data

@app.get("/sensor-data/")
def get_sensor_data(db: Session = Depends(get_db)):
    return db.query(SensorData).all()

 

 

실행

uvicorn main:app --reload

API 테스트

1. 데이터 생성

curl -X POST "http://127.0.0.1:8000/sensor-data/" -H "Content-Type: application/json" -d '{"sensor_name": "temperature", "value": 22.5}'

1. 데이터 조회

curl "http://127.0.0.1:8000/sensor-data/"

---

 

 

테이블 속성 확인

SELECT *
FROM timescaledb_information.hypertables;

 

• hypertable_schema: 하이퍼테이블이 속한 스키마 이름, 기본적으로 TimescaleDB에서 생성된 하이퍼테이블은 public 스키마
• hypertable_name: 하이퍼테이블 이름.
• owner: 하이퍼테이블의 소유자(데이터베이스 사용자) 이름, 해당 테이블을 생성한 PostgreSQL 사용자의 이름
• num_chunks: 해당 하이퍼테이블에 생성된 청크의 개수, chunk_time_interval 설정 값에 따름
• tablespaces: PostgreSQL의 테이블스페이스는 데이터를 저장할 디스크의 위치를 지정하는 기능

 

(데이터가 몇개 없어서 chunks가 1로 설정되어 있다. 데이터가 많아지면 청크 사이즈가 커지면서 자동으로 파티셔닝이 된다.)