QuestDB

# QuestDB 시계열 데이터 처리 예제

import questdb.ingress as qdb
import psycopg2
from datetime import datetime, timedelta
import random
import time

# ==========================================
# 1. ILP (InfluxDB Line Protocol)로 데이터 수집
# ==========================================

def ingest_sensor_data():
    """고속 데이터 수집 (ILP over TCP)"""
    with qdb.Sender('localhost', 9009) as sender:
        # 배치로 센서 데이터 전송
        for i in range(100000):
            sender.row(
                'sensors',
                symbols={'device_id': f'device_{i % 100}', 'location': 'seoul'},
                columns={
                    'temperature': 20.0 + random.random() * 10,
                    'humidity': 40.0 + random.random() * 30,
                    'pressure': 1013.0 + random.random() * 20
                },
                at=datetime.utcnow()
            )

            # 1000행마다 flush (성능 최적화)
            if i % 1000 == 0:
                sender.flush()

    print("100,000 rows ingested")

# ==========================================
# 2. PostgreSQL 프로토콜로 쿼리
# ==========================================

def query_with_psycopg2():
    """PostgreSQL 클라이언트로 SQL 쿼리"""
    conn = psycopg2.connect(
        host='localhost',
        port=8812,
        user='admin',
        password='quest',
        database='qdb'
    )

    with conn.cursor() as cur:
        # 테이블 생성 (시계열 최적화)
        cur.execute("""
            CREATE TABLE IF NOT EXISTS trades (
                timestamp TIMESTAMP,
                symbol SYMBOL,
                price DOUBLE,
                volume LONG
            ) timestamp(timestamp) PARTITION BY DAY WAL;
        """)

        # 최근 1시간 데이터 조회
        cur.execute("""
            SELECT * FROM trades
            WHERE timestamp > dateadd('h', -1, now())
            ORDER BY timestamp DESC
            LIMIT 100;
        """)
        rows = cur.fetchall()

        # SAMPLE BY로 시간 집계 (시계열 특화 기능)
        cur.execute("""
            SELECT
                timestamp,
                symbol,
                avg(price) as avg_price,
                sum(volume) as total_volume,
                first(price) as open,
                last(price) as close,
                max(price) as high,
                min(price) as low
            FROM trades
            WHERE timestamp > dateadd('d', -1, now())
            SAMPLE BY 1h
            ALIGN TO CALENDAR;
        """)
        ohlc_data = cur.fetchall()
        print(f"OHLC candles: {len(ohlc_data)}")

        # LATEST ON으로 각 심볼의 최신 데이터 조회
        cur.execute("""
            SELECT * FROM trades
            LATEST ON timestamp PARTITION BY symbol;
        """)
        latest = cur.fetchall()
        print(f"Latest per symbol: {latest}")

    conn.close()

# ==========================================
# 3. REST API로 데이터 조회
# ==========================================

import requests

def query_via_rest():
    """HTTP REST API로 쿼리"""
    query = """
        SELECT
            timestamp,
            device_id,
            avg(temperature) as avg_temp
        FROM sensors
        WHERE timestamp > dateadd('h', -6, now())
        SAMPLE BY 15m
    """

    response = requests.get(
        'http://localhost:9000/exec',
        params={'query': query}
    )

    if response.status_code == 200:
        data = response.json()
        print(f"Columns: {data['columns']}")
        print(f"Row count: {data['count']}")
        for row in data['dataset'][:5]:
            print(row)

# ==========================================
# 4. 데이터 보존 정책 설정
# ==========================================

def setup_retention_policy():
    """오래된 데이터 자동 삭제 설정"""
    conn = psycopg2.connect(host='localhost', port=8812, user='admin', password='quest')

    with conn.cursor() as cur:
        # 파티션 삭제 (30일 이전 데이터)
        cur.execute("""
            ALTER TABLE sensors DROP PARTITION
            WHERE timestamp < dateadd('d', -30, now());
        """)

        # 테이블 설정 변경
        cur.execute("""
            ALTER TABLE sensors SET PARAM maxUncommittedRows = 100000;
        """)

    conn.close()

# ==========================================
# 5. Grafana 대시보드용 쿼리
# ==========================================

def grafana_compatible_queries():
    """Grafana에서 사용할 시계열 쿼리 예시"""

    # 시간 범위 변수를 사용한 쿼리 (Grafana에서 $__timeFilter 대체)
    grafana_query = """
        SELECT
            timestamp as time,
            device_id,
            temperature,
            humidity
        FROM sensors
        WHERE timestamp BETWEEN '2024-01-01' AND '2024-01-02'
        SAMPLE BY $__interval
    """

    # 히트맵용 쿼리
    heatmap_query = """
        SELECT
            timestamp as time,
            floor(temperature) as temp_bucket,
            count() as count
        FROM sensors
        WHERE timestamp > dateadd('d', -7, now())
        SAMPLE BY 1h
    """

    # 알림용 집계 쿼리
    alert_query = """
        SELECT
            device_id,
            avg(temperature) as avg_temp,
            max(temperature) as max_temp
        FROM sensors
        WHERE timestamp > dateadd('m', -5, now())
        GROUP BY device_id
        HAVING avg(temperature) > 30
    """

    return grafana_query, heatmap_query, alert_query

# 실행
if __name__ == "__main__":
    ingest_sensor_data()
    query_with_psycopg2()
    query_via_rest()