TimescaleDB IoT Time Series Data Pipeline Guide

TimescaleDB Time Series Data Pipelines for IoT

TimescaleDB time series capabilities make it the ideal database for IoT data pipelines. As a PostgreSQL extension, TimescaleDB combines the reliability and ecosystem of PostgreSQL with specialized time-series optimizations — automatic partitioning, columnar compression, continuous aggregates, and data retention policies. This means your team can build high-performance IoT analytics without abandoning familiar SQL and PostgreSQL tooling.

This guide covers building a complete IoT data pipeline from sensor ingestion to real-time dashboards. We will configure hypertables, implement compression policies, create continuous aggregates for fast analytics, and set up data retention for cost-effective long-term storage.

Architecture Overview

A typical IoT pipeline ingests thousands to millions of data points per second from sensors, stores them efficiently, and makes them queryable for real-time monitoring and historical analysis:

IoT Data Pipeline Architecture

Sensors/Devices → MQTT Broker → Ingestion Service → TimescaleDB
                                      ↓                  ↓
                              (batch inserts)    ┌────────────────┐
                                                 │ Hypertables    │
                                                 │ (raw data)     │
                                                 ├────────────────┤
                                                 │ Continuous     │
                                                 │ Aggregates     │
                                                 │ (1min, 1hr)    │
                                                 ├────────────────┤
                                                 │ Compression    │
                                                 │ (older data)   │
                                                 └────────────────┘
                                                        ↓
                                                 Grafana Dashboard

TimescaleDB time series IoT data pipeline architecture — IoT data flows from sensors through ingestion into TimescaleDB hypertables

Setting Up TimescaleDB

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

-- Create the sensor readings table
CREATE TABLE sensor_readings (
    time        TIMESTAMPTZ NOT NULL,
    device_id   TEXT NOT NULL,
    location    TEXT NOT NULL,
    temperature DOUBLE PRECISION,
    humidity    DOUBLE PRECISION,
    pressure    DOUBLE PRECISION,
    battery     DOUBLE PRECISION,
    metadata    JSONB DEFAULT '{}'
);

-- Convert to a hypertable (automatic time partitioning)
SELECT create_hypertable('sensor_readings', 'time',
    chunk_time_interval => INTERVAL '1 day',
    if_not_exists => TRUE
);

-- Add space partitioning for multi-tenant/multi-device
SELECT add_dimension('sensor_readings', 'device_id',
    number_partitions => 4
);

-- Create indexes for common query patterns
CREATE INDEX idx_readings_device_time
    ON sensor_readings (device_id, time DESC);
CREATE INDEX idx_readings_location
    ON sensor_readings (location, time DESC);

High-Performance Data Ingestion

IoT pipelines require high write throughput. Moreover, batching inserts dramatically improves performance compared to individual row inserts:

import psycopg2
from psycopg2.extras import execute_values
import time
from collections import deque

class TimeseriesIngester:
    def __init__(self, dsn, batch_size=1000, flush_interval=5):
        self.conn = psycopg2.connect(dsn)
        self.conn.autocommit = False
        self.batch_size = batch_size
        self.flush_interval = flush_interval
        self.buffer = deque()
        self.last_flush = time.time()

    def insert(self, reading):
        """Buffer a reading for batch insert."""
        self.buffer.append((
            reading['time'],
            reading['device_id'],
            reading['location'],
            reading['temperature'],
            reading['humidity'],
            reading['pressure'],
            reading['battery'],
        ))

        if (len(self.buffer) >= self.batch_size or
                time.time() - self.last_flush > self.flush_interval):
            self.flush()

    def flush(self):
        """Batch insert buffered readings."""
        if not self.buffer:
            return

        batch = list(self.buffer)
        self.buffer.clear()

        with self.conn.cursor() as cur:
            execute_values(
                cur,
                """INSERT INTO sensor_readings
                   (time, device_id, location, temperature,
                    humidity, pressure, battery)
                   VALUES %s""",
                batch,
                page_size=self.batch_size
            )
        self.conn.commit()
        self.last_flush = time.time()
        print(f"Flushed {len(batch)} readings")

# Usage: processes 50,000+ inserts/second
ingester = TimeseriesIngester(
    'postgresql://user:pass@localhost/iot',
    batch_size=5000,
    flush_interval=2
)

Continuous Aggregates for Fast Analytics

Continuous aggregates automatically pre-compute time-bucketed summaries. Therefore, dashboard queries that would scan millions of rows instead read pre-materialized results:

-- 1-minute aggregates (real-time materialized view)
CREATE MATERIALIZED VIEW sensor_readings_1min
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 minute', time) AS bucket,
    device_id,
    location,
    AVG(temperature) AS avg_temp,
    MIN(temperature) AS min_temp,
    MAX(temperature) AS max_temp,
    AVG(humidity) AS avg_humidity,
    AVG(pressure) AS avg_pressure,
    AVG(battery) AS avg_battery,
    COUNT(*) AS reading_count
FROM sensor_readings
GROUP BY bucket, device_id, location
WITH NO DATA;

-- Refresh policy: auto-update aggregates
SELECT add_continuous_aggregate_policy('sensor_readings_1min',
    start_offset => INTERVAL '3 hours',
    end_offset => INTERVAL '1 minute',
    schedule_interval => INTERVAL '1 minute'
);

-- 1-hour aggregates (for historical dashboards)
CREATE MATERIALIZED VIEW sensor_readings_1hr
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 hour', bucket) AS bucket,
    device_id,
    location,
    AVG(avg_temp) AS avg_temp,
    MIN(min_temp) AS min_temp,
    MAX(max_temp) AS max_temp,
    AVG(avg_humidity) AS avg_humidity,
    SUM(reading_count) AS reading_count
FROM sensor_readings_1min
GROUP BY time_bucket('1 hour', bucket), device_id, location
WITH NO DATA;

SELECT add_continuous_aggregate_policy('sensor_readings_1hr',
    start_offset => INTERVAL '2 days',
    end_offset => INTERVAL '1 hour',
    schedule_interval => INTERVAL '1 hour'
);

Real-time IoT analytics dashboard with TimescaleDB — Continuous aggregates enable real-time dashboards over billions of data points

Compression and Data Retention

TimescaleDB columnar compression reduces storage by 90-95%. Additionally, retention policies automatically drop old data to manage costs:

-- Enable compression on the hypertable
ALTER TABLE sensor_readings SET (
    timescaledb.compress,
    timescaledb.compress_segmentby = 'device_id',
    timescaledb.compress_orderby = 'time DESC'
);

-- Auto-compress data older than 7 days
SELECT add_compression_policy('sensor_readings',
    compress_after => INTERVAL '7 days'
);

-- Retention: drop raw data older than 90 days
SELECT add_retention_policy('sensor_readings',
    drop_after => INTERVAL '90 days'
);

-- Keep 1-min aggregates for 1 year
SELECT add_retention_policy('sensor_readings_1min',
    drop_after => INTERVAL '1 year'
);

-- Keep hourly aggregates forever (no policy)

-- Check compression stats
SELECT
    pg_size_pretty(before_compression_total_bytes) AS before,
    pg_size_pretty(after_compression_total_bytes) AS after,
    round((1 - after_compression_total_bytes::numeric /
           before_compression_total_bytes) * 100, 1) AS ratio
FROM hypertable_compression_stats('sensor_readings');

When NOT to Use TimescaleDB

TimescaleDB is optimized for time-series workloads. If your data is not time-ordered — for example, a social network graph, product catalog, or document store — PostgreSQL without TimescaleDB is more appropriate. Furthermore, for extremely high cardinality metrics (millions of unique series), dedicated metrics databases like VictoriaMetrics or Mimir may be more efficient. If you need global multi-region replication with strong consistency, managed solutions like Amazon Timestream might be simpler to operate.

Database selection guide for time series data — Choose TimescaleDB when you need SQL compatibility with time-series optimizations

Key Takeaways

TimescaleDB time series capabilities extend PostgreSQL with automatic partitioning, compression, and continuous aggregates
Batch inserts achieve 50,000+ rows/second on modest hardware — sufficient for most IoT deployments
Continuous aggregates pre-compute time-bucketed summaries, reducing dashboard query times from seconds to milliseconds
Columnar compression reduces storage by 90-95%, making long-term data retention cost-effective
Retention policies automate data lifecycle management across raw, aggregated, and compressed tiers

TimescaleDB for IoT: Building High-Performance Time Series Data Pipelines