Time-Series Databases

Traditional Database Problem:

Table: sensor_readings (1 billion rows)
┌────────────┬───────────┬─────────┬───────────┐
│ timestamp  │ sensor_id │ value   │ location  │
├────────────┼───────────┼─────────┼───────────┤
│ 2024-06-15 │ sensor_1  │ 23.5    │ building_a│
│ 2024-06-15 │ sensor_2  │ 24.1    │ building_a│
│ ...        │ ...       │ ...     │ ...       │
└────────────┴───────────┴─────────┴───────────┘

Query: SELECT AVG(value) FROM sensor_readings
       WHERE timestamp >= NOW() - INTERVAL '24 hours'
       AND sensor_id = 'sensor_1'

Problem:
• Table scan of 1 billion rows (slow!)
• Index on (timestamp, sensor_id) huge
• No compression (each row stored separately)
• Query time: 30+ seconds

Time-Series Database Solution:
• Automatic partitioning by time (only scan last day)
• Columnar compression (delta encoding for values)
• Pre-aggregated rollups (hourly averages cached)
• Query time: < 100ms (300x faster!)

-- Install TimescaleDB extension
CREATE EXTENSION IF NOT EXISTS timescaledb;

-- Create regular PostgreSQL table
CREATE TABLE sensor_data (
    time        TIMESTAMPTZ NOT NULL,
    sensor_id   TEXT NOT NULL,
    temperature DOUBLE PRECISION,
    humidity    DOUBLE PRECISION,
    location    TEXT
);

-- Convert to hypertable (automatic time-based partitioning)
SELECT create_hypertable('sensor_data', 'time');

-- Create indexes for common queries
CREATE INDEX ON sensor_data (sensor_id, time DESC);
CREATE INDEX ON sensor_data (location, time DESC);

import psycopg2
from datetime import datetime, timedelta
import random

# Connect to TimescaleDB (PostgreSQL with extension)
conn = psycopg2.connect(
    dbname="timeseries_db",
    user="postgres",
    host="localhost"
)
cursor = conn.cursor()

# Simulate sensor data
sensors = ['sensor_1', 'sensor_2', 'sensor_3']
locations = ['building_a', 'building_b', 'building_c']

print("Inserting 10,000 sensor readings...")
base_time = datetime.now() - timedelta(days=7)

for i in range(10000):
    timestamp = base_time + timedelta(minutes=i)
    sensor_id = random.choice(sensors)
    location = random.choice(locations)
    temperature = round(random.uniform(18.0, 28.0), 2)
    humidity = round(random.uniform(30.0, 70.0), 2)

    cursor.execute("""
        INSERT INTO sensor_data (time, sensor_id, temperature, humidity, location)
        VALUES (%s, %s, %s, %s, %s)
    """, (timestamp, sensor_id, temperature, humidity, location))

    if (i + 1) % 1000 == 0:
        print(f"  Inserted {i + 1} records...")
        conn.commit()

conn.commit()
print("✓ Data insertion complete!")

cursor.close()
conn.close()

import psycopg2
from datetime import datetime, timedelta

conn = psycopg2.connect(dbname="timeseries_db", user="postgres")
cursor = conn.cursor()

# Query 1: Recent data (last 24 hours)
cursor.execute("""
    SELECT
        sensor_id,
        AVG(temperature) as avg_temp,
        MAX(temperature) as max_temp,
        MIN(temperature) as min_temp,
        COUNT(*) as reading_count
    FROM sensor_data
    WHERE time >= NOW() - INTERVAL '24 hours'
    GROUP BY sensor_id
    ORDER BY sensor_id
""")

print("Sensor statistics (last 24 hours):")
for sensor, avg_t, max_t, min_t, count in cursor.fetchall():
    print(f"  {sensor}: Avg={avg_t:.1f}°C, Max={max_t:.1f}°C, Min={min_t:.1f}°C ({count} readings)")

# Query 2: Time-bucket aggregation (hourly averages)
cursor.execute("""
    SELECT
        time_bucket('1 hour', time) AS hour,
        sensor_id,
        AVG(temperature) as avg_temp,
        AVG(humidity) as avg_humidity
    FROM sensor_data
    WHERE time >= NOW() - INTERVAL '7 days'
    GROUP BY hour, sensor_id
    ORDER BY hour DESC, sensor_id
    LIMIT 10
""")

print("\nHourly averages (last 7 days, most recent first):")
for hour, sensor, temp, humidity in cursor.fetchall():
    print(f"  {hour} - {sensor}: {temp:.1f}°C, {humidity:.1f}%")

cursor.close()
conn.close()

-- Enable compression on chunks older than 7 days
ALTER TABLE sensor_data SET (
    timescaledb.compress,
    timescaledb.compress_segmentby = 'sensor_id, location',
    timescaledb.compress_orderby = 'time DESC'
);

-- Add compression policy (automatic)
SELECT add_compression_policy('sensor_data', INTERVAL '7 days');

-- Manually compress specific chunks (for demo)
SELECT compress_chunk(i)
FROM show_chunks('sensor_data', older_than => INTERVAL '7 days') i;

import psycopg2

conn = psycopg2.connect(dbname="timeseries_db", user="postgres")
cursor = conn.cursor()

# Check compression statistics
cursor.execute("""
    SELECT
        pg_size_pretty(before_compression_total_bytes) as uncompressed,
        pg_size_pretty(after_compression_total_bytes) as compressed,
        ROUND(100 - (after_compression_total_bytes::numeric /
              before_compression_total_bytes::numeric * 100), 2) as compression_ratio
    FROM hypertable_compression_stats('sensor_data')
""")

result = cursor.fetchone()
if result:
    uncompressed, compressed, ratio = result
    print("Compression Statistics:")
    print(f"  Uncompressed size: {uncompressed}")
    print(f"  Compressed size: {compressed}")
    print(f"  Space saved: {ratio}%")
else:
    print("No compression data available yet")

cursor.close()
conn.close()

-- Automatically drop data older than 90 days
SELECT add_retention_policy('sensor_data', INTERVAL '90 days');

-- Check retention policy
SELECT * FROM timescaledb_information.jobs
WHERE proc_name = 'policy_retention';

InfluxDB Data Structure:

measurement (like a table)
│
├─ tags (indexed, metadata)
│  ├─ sensor_id = "sensor_1"
│  ├─ location = "building_a"
│  └─ region = "us-east"
│
├─ fields (actual data, not indexed)
│  ├─ temperature = 23.5
│  └─ humidity = 52.3
│
└─ timestamp = 2024-06-15T10:30:00Z

Example data point:
┌─────────────────────────────────────────────────┐
│ Measurement: sensor_readings                    │
├─────────────────────────────────────────────────┤
│ Tags:                                           │
│   sensor_id=sensor_1, location=building_a       │
│ Fields:                                         │
│   temperature=23.5, humidity=52.3               │
│ Timestamp: 2024-06-15T10:30:00Z                 │
└─────────────────────────────────────────────────┘

Key Concepts:
• Measurement = table name
• Tags = indexed dimensions (WHERE filters)
• Fields = actual values (SELECT columns)
• Timestamp = automatically indexed

from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
from datetime import datetime, timedelta
import random

# Connect to InfluxDB
client = InfluxDBClient(
    url="http://localhost:8086",
    token="your-token-here",
    org="my-org"
)

write_api = client.write_api(write_options=SYNCHRONOUS)
bucket = "sensor_data"

# Write sensor readings
print("Writing sensor data to InfluxDB...")
sensors = ['sensor_1', 'sensor_2', 'sensor_3']
locations = ['building_a', 'building_b']

base_time = datetime.utcnow() - timedelta(hours=24)

for i in range(1000):
    timestamp = base_time + timedelta(minutes=i)

    for sensor in sensors:
        for location in locations:
            point = Point("sensor_readings") \
                .tag("sensor_id", sensor) \
                .tag("location", location) \
                .field("temperature", round(random.uniform(18.0, 28.0), 2)) \
                .field("humidity", round(random.uniform(30.0, 70.0), 2)) \
                .time(timestamp)

            write_api.write(bucket=bucket, record=point)

    if (i + 1) % 100 == 0:
        print(f"  Written {(i + 1) * len(sensors) * len(locations)} points...")

print("✓ Data written successfully!")
client.close()

from influxdb_client import InfluxDBClient

client = InfluxDBClient(
    url="http://localhost:8086",
    token="your-token-here",
    org="my-org"
)

query_api = client.query_api()

# Query 1: Average temperature by sensor (last 24 hours)
query = '''
from(bucket: "sensor_data")
  |> range(start: -24h)
  |> filter(fn: (r) => r._measurement == "sensor_readings")
  |> filter(fn: (r) => r._field == "temperature")
  |> group(columns: ["sensor_id"])
  |> mean()
'''

tables = query_api.query(query)

print("Average temperature by sensor (last 24h):")
for table in tables:
    for record in table.records:
        sensor = record.values.get("sensor_id")
        avg_temp = record.get_value()
        print(f"  {sensor}: {avg_temp:.2f}°C")

# Query 2: Downsampled data (hourly averages)
query = '''
from(bucket: "sensor_data")
  |> range(start: -7d)
  |> filter(fn: (r) => r._measurement == "sensor_readings")
  |> filter(fn: (r) => r.sensor_id == "sensor_1")
  |> filter(fn: (r) => r._field == "temperature")
  |> aggregateWindow(every: 1h, fn: mean, createEmpty: false)
  |> limit(n: 10)
'''

tables = query_api.query(query)

print("\nHourly temperature averages for sensor_1:")
for table in tables:
    for record in table.records:
        time = record.get_time()
        temp = record.get_value()
        print(f"  {time}: {temp:.2f}°C")

client.close()

# Create downsampling task (runs every hour)
# Reduces high-resolution data to hourly averages

from influxdb_client import InfluxDBClient
from influxdb_client.client.tasks_api import TasksApi

client = InfluxDBClient(url="http://localhost:8086", token="your-token-here", org="my-org")
tasks_api = TasksApi(client)

# Task: Calculate hourly averages and store in separate bucket
task_flux = '''
option task = {name: "downsample_hourly", every: 1h}

from(bucket: "sensor_data")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "sensor_readings")
  |> aggregateWindow(every: 1h, fn: mean, createEmpty: false)
  |> to(bucket: "sensor_data_hourly")
'''

task = tasks_api.create_task_every(
    name="downsample_hourly",
    flux=task_flux,
    every="1h",
    organization="my-org"
)

print(f"Created downsampling task: {task.name}")
print(f"  Runs every: {task.every}")
print("  Purpose: Reduce raw data to hourly averages")

client.close()

from influxdb_client import InfluxDBClient
from influxdb_client.client.buckets_api import BucketsApi, RetentionRules

client = InfluxDBClient(url="http://localhost:8086", token="your-token-here", org="my-org")
buckets_api = BucketsApi(client)

# Create bucket with 30-day retention for raw data
retention_rules = RetentionRules(type="expire", every_seconds=30 * 24 * 60 * 60)  # 30 days

raw_bucket = buckets_api.create_bucket(
    bucket_name="sensor_data_raw",
    retention_rules=retention_rules,
    org="my-org"
)

print("Created buckets with retention policies:")
print(f"  Raw data (sensor_data_raw): 30 days retention")

# Create bucket with 1-year retention for downsampled data
retention_rules_long = RetentionRules(type="expire", every_seconds=365 * 24 * 60 * 60)  # 1 year

hourly_bucket = buckets_api.create_bucket(
    bucket_name="sensor_data_hourly",
    retention_rules=retention_rules_long,
    org="my-org"
)

print(f"  Downsampled data (sensor_data_hourly): 1 year retention")
print("\n✓ Old data automatically deleted after retention period!")

client.close()

Prometheus Metrics:

Metric name + labels = time series

Example:
http_requests_total{method="GET", endpoint="/api/users", status="200"} 12345

Components:
• http_requests_total = metric name
• {method="GET", endpoint="/api/users", status="200"} = labels
• 12345 = value at this timestamp

Metric Types:
1. Counter: Cumulative value (only increases)
   - http_requests_total
   - errors_total

2. Gauge: Value that can go up or down
   - memory_usage_bytes
   - active_connections

3. Histogram: Distribution of values
   - request_duration_seconds
   - response_size_bytes

4. Summary: Similar to histogram (percentiles)
   - api_latency_seconds

from prometheus_client import start_http_server, Counter, Gauge, Histogram
import random
import time

# Define metrics
request_counter = Counter(
    'http_requests_total',
    'Total HTTP requests',
    ['method', 'endpoint', 'status']
)

active_users = Gauge(
    'active_users',
    'Number of active users'
)

request_duration = Histogram(
    'request_duration_seconds',
    'Request duration in seconds',
    ['endpoint']
)

# Simulate application metrics
def simulate_traffic():
    """Simulate web application traffic"""
    endpoints = ['/api/users', '/api/products', '/api/orders']
    methods = ['GET', 'POST']
    statuses = ['200', '404', '500']

    while True:
        # Increment request counter
        endpoint = random.choice(endpoints)
        method = random.choice(methods)
        status = random.choice(statuses)
        request_counter.labels(method=method, endpoint=endpoint, status=status).inc()

        # Update active users gauge
        active_users.set(random.randint(100, 500))

        # Record request duration
        duration = random.uniform(0.01, 2.0)
        request_duration.labels(endpoint=endpoint).observe(duration)

        time.sleep(1)  # Wait 1 second

if __name__ == '__main__':
    # Start metrics server on port 8000
    start_http_server(8000)
    print("Metrics server started on port 8000")
    print("Prometheus can scrape: http://localhost:8000/metrics")

    # Simulate traffic
    simulate_traffic()

from prometheus_api_client import PrometheusConnect
import pandas as pd

# Connect to Prometheus server
prom = PrometheusConnect(url="http://localhost:9090", disable_ssl=True)

# Query 1: Current request rate (requests per second)
query = 'rate(http_requests_total[5m])'
result = prom.custom_query(query)

print("Current request rate (requests/sec):")
for item in result:
    metric = item['metric']
    value = float(item['value'][1])
    print(f"  {metric['endpoint']} ({metric['method']}): {value:.2f} req/s")

# Query 2: 95th percentile request duration
query = 'histogram_quantile(0.95, rate(request_duration_seconds_bucket[5m]))'
result = prom.custom_query(query)

print("\n95th percentile request duration:")
for item in result:
    endpoint = item['metric'].get('endpoint', 'unknown')
    p95 = float(item['value'][1])
    print(f"  {endpoint}: {p95:.3f}s")

# Query 3: Error rate (percentage)
query = '''
  sum(rate(http_requests_total{status=~"5.."}[5m])) /
  sum(rate(http_requests_total[5m])) * 100
'''
result = prom.custom_query(query)

if result:
    error_rate = float(result[0]['value'][1])
    print(f"\nCurrent error rate: {error_rate:.2f}%")

# prometheus.yml configuration

global:
  scrape_interval: 15s      # How often to scrape targets
  evaluation_interval: 15s  # How often to evaluate rules

# Scrape configurations
scrape_configs:
  # Job 1: Python application metrics
  - job_name: 'python_app'
    static_configs:
      - targets: ['localhost:8000']
        labels:
          environment: 'production'
          app: 'sensor_api'

  # Job 2: Node exporter (system metrics)
  - job_name: 'node'
    static_configs:
      - targets: ['localhost:9100']

  # Job 3: PostgreSQL metrics
  - job_name: 'postgres'
    static_configs:
      - targets: ['localhost:9187']

# Storage retention (configured via CLI flag, not in this file):
#   prometheus --storage.tsdb.retention.time=15d

Typical Multi-Tier Retention:

Tier 1: Raw Data (Full Resolution)
┌─────────────────────────────────────────┐
│ Retention: 7 days                       │
│ Granularity: 1 second                   │
│ Storage: 100 GB                         │
│ Use: Recent detailed analysis           │
└─────────────────────────────────────────┘

Tier 2: Hourly Aggregates
┌─────────────────────────────────────────┐
│ Retention: 90 days                      │
│ Granularity: 1 hour                     │
│ Storage: 5 GB (95% reduction!)          │
│ Use: Weekly/monthly reports             │
└─────────────────────────────────────────┘

Tier 3: Daily Aggregates
┌─────────────────────────────────────────┐
│ Retention: 2 years                      │
│ Granularity: 1 day                      │
│ Storage: 500 MB (99.5% reduction!)      │
│ Use: Long-term trend analysis           │
└─────────────────────────────────────────┘

Benefits:
• Keep detailed data when it matters (recent)
• Reduce storage costs (compress old data)
• Fast queries (pre-aggregated)
• Compliance (retain summaries long-term)

-- Create continuous aggregate for hourly data
CREATE MATERIALIZED VIEW sensor_data_hourly
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 hour', time) AS hour,
    sensor_id,
    location,
    AVG(temperature) as avg_temperature,
    MAX(temperature) as max_temperature,
    MIN(temperature) as min_temperature,
    AVG(humidity) as avg_humidity,
    COUNT(*) as reading_count
FROM sensor_data
GROUP BY hour, sensor_id, location;

-- Create continuous aggregate for daily data
CREATE MATERIALIZED VIEW sensor_data_daily
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 day', time) AS day,
    sensor_id,
    location,
    AVG(temperature) as avg_temperature,
    MAX(temperature) as max_temperature,
    MIN(temperature) as min_temperature,
    AVG(humidity) as avg_humidity,
    COUNT(*) as reading_count
FROM sensor_data
GROUP BY day, sensor_id, location;

-- Retention policies
SELECT add_retention_policy('sensor_data', INTERVAL '7 days');         -- Raw: 7 days
SELECT add_retention_policy('sensor_data_hourly', INTERVAL '90 days'); -- Hourly: 90 days
SELECT add_retention_policy('sensor_data_daily', INTERVAL '2 years');  -- Daily: 2 years

-- Compression policies
SELECT add_compression_policy('sensor_data', INTERVAL '1 day');
SELECT add_compression_policy('sensor_data_hourly', INTERVAL '7 days');

import psycopg2
from datetime import datetime, timedelta

conn = psycopg2.connect(dbname="timeseries_db", user="postgres")
cursor = conn.cursor()

# Query recent data (uses raw table - high resolution)
cursor.execute("""
    SELECT
        time_bucket('5 minutes', time) as interval,
        AVG(temperature) as avg_temp
    FROM sensor_data
    WHERE time >= NOW() - INTERVAL '6 hours'
        AND sensor_id = 'sensor_1'
    GROUP BY interval
    ORDER BY interval DESC
    LIMIT 10
""")

print("Recent data (5-minute intervals, last 6 hours):")
for interval, temp in cursor.fetchall():
    print(f"  {interval}: {temp:.2f}°C")

# Query older data (uses hourly aggregate - pre-computed)
cursor.execute("""
    SELECT
        hour,
        avg_temperature
    FROM sensor_data_hourly
    WHERE hour >= NOW() - INTERVAL '30 days'
        AND hour < NOW() - INTERVAL '7 days'
        AND sensor_id = 'sensor_1'
    ORDER BY hour DESC
    LIMIT 10
""")

print("\nOlder data (hourly averages, 7-30 days ago):")
for hour, temp in cursor.fetchall():
    print(f"  {hour}: {temp:.2f}°C")

# Query historical data (uses daily aggregate)
cursor.execute("""
    SELECT
        day,
        avg_temperature,
        max_temperature,
        min_temperature
    FROM sensor_data_daily
    WHERE day >= NOW() - INTERVAL '1 year'
        AND sensor_id = 'sensor_1'
    ORDER BY day DESC
    LIMIT 10
""")

print("\nHistorical data (daily averages, last year):")
for day, avg_t, max_t, min_t in cursor.fetchall():
    print(f"  {day}: Avg={avg_t:.1f}°C, Max={max_t:.1f}°C, Min={min_t:.1f}°C")

cursor.close()
conn.close()