TimescaleDB Time-Series Importer

TimescaleDB Time-Series Importer

Problem Statement

Ingesting high-volume time-series data (IoT sensor readings, application metrics, financial ticks) into a database is a common but surprisingly fiddly task. Naive approaches like INSERT INTO ... VALUES (...) per row collapse under load, and even batched inserts often fail because a single malformed timestamp in a 10,000-row batch rolls back the entire transaction. Meanwhile, TimescaleDB's hypertables are optimized for append-heavy workloads — but only if the importer respects chunk boundaries, uses COPY or multi-row INSERT, and commits at a sane cadence.

This script solves that by streaming a CSV file into a TimescaleDB hypertable in configurable batches, with automatic type coercion for timestamps and numerics, retry-on-transient-failure logic, and a --dry-run mode so you can validate a file before touching production. It's designed to be dropped into a cron job, an Airflow DAG, or run interactively by a sysadmin.

How It Works

The script reads the CSV with csv.DictReader so column order doesn't matter — it maps named columns from the file to named columns in the target table. It pulls the connection string from TIMESCALE_DSN (or standard PG* env vars), opens a single psycopg2 connection, and iterates over the file in batches of --chunk rows. Each batch is inserted using psycopg2.extras.execute_values, which compiles a single multi-row INSERT ... VALUES (...), (...), ... statement — dramatically faster than per-row execution and safer than COPY when you need per-row error isolation.

The design uses three key patterns: (1) generator-based batching so memory stays flat regardless of file size, (2) type coercion hooks for timestamp and numeric columns, and (3) exponential backoff retries on psycopg2.OperationalError (connection drops, lock timeouts) while letting DataError/IntegrityError fail fast on bad data. On any batch failure, the whole batch is rolled back and the script exits with a non-zero code, so upstream orchestrators can retry the whole file idempotently (assuming you use ON CONFLICT DO NOTHING or a unique constraint).

Full Script

#!/usr/bin/env python3
"""
TimescaleDB Time-Series Importer

Streams a CSV file into a TimescaleDB hypertable in configurable batches.
Handles timestamp/numeric coercion, retries on transient connection errors,
and supports a dry-run mode for validation.

Typical usage:
    python3 timescale_import.py --table metrics --input metrics.csv \
        --time-col timestamp --chunk 10000
"""

from __future__ import annotations

import argparse
import csv
import logging
import os
import sys
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Iterable, Iterator, Sequence

import psycopg2
from psycopg2 import OperationalError, sql
from psycopg2.extras import execute_values

# ---------------------------------------------------------------------------
# Configuration (env vars with sensible defaults)
# ---------------------------------------------------------------------------
TIMESCALE_DSN = os.environ.get(
    "TIMESCALE_DSN",
    "postgresql://postgres:postgres@localhost:5432/postgres",
)
CONNECT_TIMEOUT = int(os.environ.get("TIMESCALE_CONNECT_TIMEOUT", "10"))
STATEMENT_TIMEOUT_MS = int(os.environ.get("TIMESCALE_STMT_TIMEOUT_MS", "60000"))
MAX_RETRIES = int(os.environ.get("TIMESCALE_MAX_RETRIES", "3"))
RETRY_BACKOFF_BASE = float(os.environ.get("TIMESCALE_RETRY_BACKOFF", "1.5"))

logging.basicConfig(
    level=os.environ.get("LOG_LEVEL", "INFO").upper(),
    format="%(asctime)s %(levelname)-7s %(message)s",
)
log = logging.getLogger("timescale_import")


# ---------------------------------------------------------------------------
# Data structures
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class ImportStats:
    """Counters reported at the end of a run."""
    rows_read: int = 0
    rows_inserted: int = 0
    batches: int = 0
    elapsed_s: float = 0.0


# ---------------------------------------------------------------------------
# CSV streaming
# ---------------------------------------------------------------------------
def stream_rows(path: str) -> Iterator[dict[str, str]]:
    """Yield CSV rows as dicts. Skips blank lines silently."""
    with open(path, newline="", encoding="utf-8") as fh:
        reader = csv.DictReader(fh)
        if reader.fieldnames is None:
            raise ValueError(f"CSV file {path!r} appears to be empty or has no header")
        for row in reader:
            # csv.DictReader yields None for missing trailing fields; normalize.
            yield {k: (v if v is not None else "") for k, v in row.items()}


def batched(iterable: Iterable[Any], size: int) -> Iterator[list[Any]]:
    """Group an iterable into lists of at most `size` items."""
    batch: list[Any] = []
    for item in iterable:
        batch.append(item)
        if len(batch) >= size:
            yield batch
            batch = []
    if batch:
        yield batch


# ---------------------------------------------------------------------------
# Type coercion
# ---------------------------------------------------------------------------
def coerce_timestamp(value: str) -> datetime:
    """
    Parse common timestamp formats. ISO-8601 is preferred; a few fallbacks
    are included because real-world CSVs are messy.
    """
    if not value:
        raise ValueError("empty timestamp")
    # Normalize 'Z' suffix which datetime.fromisoformat <3.11 can't parse.
    normalized = value.replace("Z", "+00:00") if value.endswith("Z") else value
    try:
        dt = datetime.fromisoformat(normalized)
    except ValueError:
        # Fallback: epoch seconds (int or float).
        try:
            dt = datetime.fromtimestamp(float(value), tz=timezone.utc)
        except (ValueError, OSError) as exc:
            raise ValueError(f"unparseable timestamp: {value!r}") from exc
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt


def coerce_value(value: str, pg_type: str) -> Any:
    """Coerce a CSV string into the Python type psycopg2 expects for `pg_type`."""
    if value == "":
        return None
    t = pg_type.lower()
    if t.startswith("timestamp") or t == "timestamptz" or t == "date":
        return coerce_timestamp(value)
    if t in ("int", "int2", "int4", "int8", "integer", "bigint", "smallint"):
        return int(value)
    if t in ("float4", "float8", "numeric", "decimal", "real", "double precision"):
        return float(value)
    if t in ("bool", "boolean"):
        return value.strip().lower() in ("1", "t", "true", "yes", "y")
    return value  # text, varchar, jsonb, etc. — pass through


# ---------------------------------------------------------------------------
# Database helpers
# ---------------------------------------------------------------------------
def connect(dsn: str) -> psycopg2.extensions.connection:
    """Open a connection with sane timeouts for an ETL workload."""
    conn = psycopg2.connect(dsn, connect_timeout=CONNECT_TIMEOUT)
    conn.autocommit = False
    with conn.cursor() as cur:
        cur.execute(f"SET statement_timeout = {STATEMENT_TIMEOUT_MS}")
    return conn


def fetch_column_types(
    conn: psycopg2.extensions.connection, table: str
) -> dict[str, str]:
    """Return {column_name: pg_type} for the target table, in ordinal order."""
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT column_name, data_type
            FROM information_schema.columns
            WHERE table_schema = current_schema()
              AND table_name = %s
            ORDER BY ordinal_position
            """,
            (table,),
        )
        rows = cur.fetchall()
    if not rows:
        raise ValueError(f"table {table!r} not found in current schema")
    return {name: dtype for name, dtype in rows}


def insert_batch(
    conn: psycopg2.extensions.connection,
    table: str,
    columns: Sequence[str],
    rows: Sequence[Sequence[Any]],
) -> int:
    """
    Insert one batch using a single multi-row INSERT. Returns rows affected.
    Uses execute_values for a compiled VALUES list — much faster than
    executemany and avoids the COPY privilege requirements.
    """
    if not rows:
        return 0
    col_sql = sql.SQL(", ").join(sql.Identifier(c) for c in columns)
    query = sql.SQL(
        "INSERT INTO {table} ({cols}) VALUES %s ON CONFLICT DO NOTHING"
    ).format(table=sql.Identifier(table), cols=col_sql)
    with conn.cursor() as cur:
        execute_values(cur, query.as_string(conn), rows, page_size=len(rows))
        affected = cur.rowcount
    conn.commit()
    return affected


# ---------------------------------------------------------------------------
# Retry wrapper
# ---------------------------------------------------------------------------
def with_retries(fn, *args, **kwargs):
    """
    Run `fn` up to MAX_RETRIES times on transient OperationalError.
    Data/integrity errors propagate immediately — retrying bad data is futile.
    """
    last_exc: Exception | None = None
    for attempt in range(1, MAX_RETRIES + 1):
        try:
            return fn(*args, **kwargs)
        except OperationalError as exc:
            last_exc = exc
            delay = RETRY_BACKOFF_BASE ** attempt
            log.warning(
                "OperationalError on attempt %d/%d: %s — retrying in %.1fs",
                attempt, MAX_RETRIES, exc, delay,
            )
            time.sleep(delay)
    assert last_exc is not None
    raise last_exc


# ---------------------------------------------------------------------------
# Main import loop
# ---------------------------------------------------------------------------
def run_import(
    dsn: str,
    table: str,
    input_path: str,
    time_col: str,
    chunk_size: int,
    dry_run: bool,
) -> ImportStats:
    start = time.monotonic()
    rows_read = rows_inserted = batches = 0

    conn = None if dry_run else with_retries(connect, dsn)
    try:
        if conn is not None:
            col_types = fetch_column_types(conn, table)
        else:
            # Dry-run: infer types loosely from the first row.
            first = next(stream_rows(input_path), None)
            if first is None:
                raise ValueError("input CSV has no data rows")
            col_types = {k: "text" for k in first}

        if time_col not in col_types:
            raise ValueError(
                f"time column {time_col!r} not found in table {table!r}; "
                f"available: {sorted(col_types)}"
            )

        # Only insert columns present in both CSV and target table.
        target_cols = list(col_types.keys())

        for batch in batched(stream_rows(input_path), chunk_size):
            batches += 1
            rows_read += len(batch)
            prepared: list[tuple[Any, ...]] = []
            for row in batch:
                try:
                    prepared.append(
                        tuple(
                            coerce_value(row.get(c, ""), col_types[c])
                            for c in target_cols
                        )
                    )
                except (ValueError, TypeError) as exc:
                    raise ValueError(
                        f"row {rows_read} failed coercion: {exc} | row={row}"
                    ) from exc

            if dry_run:
                log.info("DRY-RUN batch %d: %d rows validated", batches, len(prepared))
                rows_inserted += len(prepared)
                continue

            assert conn is not None
            affected = with_retries(
                insert_batch, conn, table, target_cols, prepared
            )
            rows_inserted += affected
            log.info(
                "batch %d: read=%d inserted=%d (cumulative %d)",
                batches, len(prepared), affected, rows_inserted,
            )
    finally:
        if conn is not None:
            conn.close()

    return ImportStats(
        rows_read=rows_read,
        rows_inserted=rows_inserted,
        batches=batches,
        elapsed_s=time.monotonic() - start,
    )


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
    p = argparse.ArgumentParser(
        description="Import a CSV file into a TimescaleDB hypertable in batches."
    )
    p.add_argument("--table", required=True, help="Target hypertable name")
    p.add_argument("--input", required=True, help="Path to input CSV file")
    p.add_argument(
        "--time-col", required=True,
        help="Name of the timestamp column (validated against the table schema)",
    )
    p.add_argument(
        "--chunk", type=int, default=10_000,
        help="Rows per batch (default: 10000)",
    )
    p.add_argument(
        "--dry-run", action="store_true",
        help="Validate and coerce rows without writing to the database",
    )
    p.add_argument(
        "--dsn", default=TIMESCALE_DSN,
        help="PostgreSQL DSN (overrides TIMESCALE_DSN env var)",
    )
    return p.parse_args(argv)


def main(argv: Sequence[str] | None = None) -> int:
    args = parse_args(argv)
    if args.chunk <= 0:
        log.error("--chunk must be positive, got %d", args.chunk)
        return 2

    try:
        stats = run_import(
            dsn=args.dsn,
            table=args.table,
            input_path=args.input,
            time_col=args.time_col,
            chunk_size=args.chunk,
            dry_run=args.dry_run,
        )
    except FileNotFoundError:
        log.error("input file not found: %s", args.input)
        return 1
    except (ValueError, psycopg2.Error) as exc:
        log.error("import failed: %s", exc)
        return 1

    log.info(
        "done: read=%d inserted=%d batches=%d elapsed=%.2fs",
        stats.rows_read, stats.rows_inserted, stats.batches, stats.elapsed_s,
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())

Configuration

Usage Examples

1. Basic import into a metrics hypertable:

$ export TIMESCALE_DSN="postgresql://ingest:pw@tsdb.internal:5432/observability"
$ python3 timescale_import.py \
    --table metrics \
    --input metrics.csv \
    --time-col timestamp \
    --chunk 10000
2024-06-01 09:14:02 INFO    batch 1: read=10000 inserted=10000 (cumulative 10000)
2024-06-01 09:14:04 INFO    batch 2: read=10000 inserted=10000 (cumulative 20000)
2024-06-01 09:14:05 INFO    done: read=20000 inserted=20000 batches=2 elapsed=3.41s

2. Validate a file in CI before deploying:

$ python3 timescale_import.py \
    --table metrics --input metrics.csv --time-col timestamp --dry-run
2024-06-01 09:15:11 INFO    DRY-RUN batch 1: 10000 rows validated
2024-06-01 09:15:12 INFO    DRY-RUN batch 2: 10000 rows validated
2024-06-01 09:15:12 INFO    done: read=20000 inserted=20000 batches=2 elapsed=0.87s

3. Small batches for a memory-constrained edge box:

$ TIMESCALE_MAX_RETRIES=5 python3 timescale_import.py \
    --table sensor_readings --input /var/spool/readings.csv \
    --time-col recorded_at --chunk 2000

Error Handling

• Missing file → FileNotFoundError caught in main, logged, exit code 1.
• Bad timestamp / numeric value → coerce_value raises ValueError with the offending row number and raw value; the batch is abandoned and the process exits 1. Because insert_batch commits per batch, earlier batches are already durable — you can fix the CSV and re-run with ON CONFLICT DO NOTHING to skip duplicates.
• Unknown time column → validated against information_schema.columns before any row is read.
• Transient DB errors (OperationalError: connection reset, lock_timeout, deadlock detected) → retried up to TIMESCALE_MAX_RETRIES with exponential backoff. Each retry re-runs the whole batch, which is safe because the insert is idempotent under ON CONFLICT DO NOTHING.
• Data errors (DataError, IntegrityError) → not retried; they indicate a schema mismatch or constraint violation that will fail identically on retry.
• Statement timeout → a batch exceeding TIMESCALE_STMT_TIMEOUT_MS aborts with QueryCanceled; lower --chunk to fix.

Extending Ideas

• Add --upsert mode that switches ON CONFLICT DO NOTHING to ON CONFLICT (time, tags) DO UPDATE SET value = EXCLUDED.value, useful for late-arriving corrections.
• Swap execute_values for COPY FROM STDIN (psycopg2.copy_expert) for 2–5× throughput on very large files; you lose per-row ON CONFLICT handling, so pair it with a staging table and INSERT ... SELECT.
• Parallelize with a process pool keyed on byte ranges of the CSV — each worker opens its own connection, and TimescaleDB's per-chunk locking keeps contention low.
• Emit Prometheus metrics (rows_inserted_total, batch_duration_seconds) by dropping in prometheus_client and exposing an HTTP endpoint, so your Grafana dashboard shows ingestion health alongside the data itself.