Geospatial Time-Series Ingestion Pipeline
Ingest 100k+ GPS records/sec into TimescaleDB with PostGIS spatial queries.
Introduction
30 Second Summary
Every ride-sharing app, delivery tracker, and fleet management system faces the same problem. Millions of GPS coordinates flood in every second, and each one needs to be stored, indexed, and queryable in real time.
In this project, you will build a high-throughput geospatial time-series ingestion pipeline that simulates 500 GPS-tracked vehicles, streams their positions into a database at 100,000+ rows per second, and runs spatial-temporal queries to analyze movement patterns.
What You'll Build
A complete ingestion pipeline combining TimescaleDB's time-series partitioning with PostGIS spatial indexing, all running locally via Docker.
By the end of this project, you'll have:
- A Docker Compose stack running TimescaleDB with PostGIS for combined time-series and spatial workloads.
- A Python ingestion script using the PostgreSQL COPY protocol to stream 200,000 simulated GPS records at high throughput.
- Working spatial-temporal queries that combine PostGIS distance filtering with TimescaleDB time_bucket aggregations to answer questions like "which vehicles were within 2km of this location in the last hour?"
- Secret Mission: Create a continuous aggregate with H3 hexagonal grid cells for pre-computed vehicle density heatmaps.
Are there any prerequisites?
You should have basic SQL knowledge and basic Python familiarity. You also need Docker Desktop installed on your machine. The project will guide you through setting up everything else.
Set Up TimescaleDB with PostGIS
Every fleet tracking system needs a database that can handle both time-stamped data and spatial queries efficiently. In this step, you'll spin up that database infrastructure using Docker Compose.
TimescaleDB extends PostgreSQL with automatic time-based partitioning for high-speed ingestion. PostGIS adds geospatial query capabilities. Together, they let you store and query timestamped GPS coordinates in a single database.
In this step, get ready to:
- Confirm Docker Desktop is running on your machine.
- Create a Docker Compose configuration for TimescaleDB with PostGIS.
- Start the database container and verify both extensions are available.
Confirm Docker is running
With Docker Desktop already running, confirm it is responsive before you depend on it for the rest of the project.
- Verify Docker is responding by running this command:
docker --versionYou should see a version number printed in the output, confirming Docker is ready.
Don't see a version number?
Make sure Docker Desktop is running. Check for the whale icon in your menu bar (macOS) or system tray (Windows).
help me troubleshoot Docker not responding on my machine
Create the Docker Compose file
Docker Compose lets you define your database container, its configuration, and persistent storage in a single YAML file. You'll use the timescale/timescaledb-ha:pg17 image, which bundles TimescaleDB and PostGIS together.
- Create a new file called docker-compose.yml in your project folder and paste the following content:
services:
timescaledb:
image: timescale/timescaledb-ha:pg17
container_name: tsdb-geo
ports:
- "5432:5432"
environment:
POSTGRES_PASSWORD: password
POSTGRES_DB: fleet_tracking
volumes:
- tsdb_data:/home/postgres/pgdata/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U postgres"]
interval: 5s
timeout: 5s
retries: 5
volumes:
tsdb_data:What does this configuration do?
- image: timescale/timescaledb-ha:pg17 pulls the TimescaleDB High Availability image, which includes PostgreSQL 17, TimescaleDB, and PostGIS out of the box.
- container_name: tsdb-geo gives the container a fixed name so you can reference it easily in later commands.
- POSTGRES_DB: fleet_tracking creates a database called fleet_tracking automatically on first start.
- volumes maps a named volume to the container's data directory so your data persists across restarts.
- healthcheck uses pg_isready to poll the database every 5 seconds, confirming it is accepting connections.
- Save docker-compose.yml.
- Confirm the file exists by running:
lsYou should see docker-compose.yml in the output.
Don't see docker-compose.yml?
Make sure you saved the file in your project folder, not a parent or child directory. Run pwd to check which directory your terminal is in.
help me figure out why my docker-compose.yml is not showing up
✔️ Awesome, I've got everything!
Great. Double check you've saved your file, then move on to the next substep.
ⓧ I'd like to double check the full code
services:
timescaledb:
image: timescale/timescaledb-ha:pg17
container_name: tsdb-geo
ports:
- "5432:5432"
environment:
POSTGRES_PASSWORD: password
POSTGRES_DB: fleet_tracking
volumes:
- tsdb_data:/home/postgres/pgdata/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U postgres"]
interval: 5s
timeout: 5s
retries: 5
volumes:
tsdb_data:Start the container and verify extensions
Now you'll bring the container up and confirm that both TimescaleDB and PostGIS are available inside the database.
- Start the TimescaleDB container in the background by running:
docker compose up -dYou should see output indicating the container and network were created. The first run may take a minute as Docker pulls the image.
What does the -d flag do?
The -d flag runs the container in detached mode. This means it starts in the background and gives you your terminal back instead of streaming logs.
Seeing a port conflict error?
If you see a message about port 5432 already being allocated, another service is using that port. Stop the other service, or change the host port in docker-compose.yml to 5433:5432.
help me resolve a port conflict with my TimescaleDB container
- Wait about 10 seconds for the database to initialize, then verify the container is healthy by running:
docker psYou should see a container named tsdb-geo with a status that includes healthy in the output.
Container not showing as healthy?
If the status shows starting, wait another 10-15 seconds and run docker ps again. The healthcheck polls every 5 seconds and needs a few successful checks before marking the container healthy.
help me troubleshoot my TimescaleDB container not becoming healthy
- Confirm both the timescaledb and postgis extensions are available by running this query inside the container:
docker exec -it tsdb-geo psql -U postgres -d fleet_tracking -c "SELECT name FROM pg_available_extensions WHERE name IN ('timescaledb', 'postgis');"You should see both timescaledb and postgis listed in the query results.
What is this command doing?
- docker exec -it tsdb-geo runs a command inside the running container named tsdb-geo.
- psql -U postgres -d fleet_tracking connects to the fleet_tracking database as the postgres user.
- -c passes a single SQL query and exits. The query checks pg_available_extensions for the two extensions your project needs.
Not seeing both extensions?
If you get a connection refused error, the database might not be ready yet. Run docker ps and wait until the container status shows healthy before retrying.
If postgis is missing, verify you are using the timescale/timescaledb-ha:pg17 image. The plain timescale/timescaledb image does not include PostGIS.
help me figure out why PostGIS is missing from my TimescaleDB container
Your database is up and running with both time-series and geospatial capabilities. Next up, you'll create the Python environment and design the schema for vehicle telemetry data.
Set Up Python Environment and Create the Schema
Your TimescaleDB container is running and ready to accept connections. Now you need a way to talk to it from Python.
In this step, you will set up a Python environment with the psycopg database adapter, then write a schema setup script that creates a hypertable with geospatial columns and indexes.
In this step, get ready to:
- Create a Python virtual environment and install the database adapter.
- Write a schema setup script that creates a hypertable with geospatial indexing.
- Run the script and verify the hypertable was created.
Create a Python virtual environment and install psycopg
The ingestion script you will write later needs a Python library to communicate with PostgreSQL. You will use psycopg (version 3), which provides a modern async-capable adapter with a high-performance COPY protocol interface.
- Check that Python 3 is installed by running this command:
python3 --version✔️ I see a version number starting with 3
Python 3 is ready. Move on to creating the virtual environment below.
ⓧ Command not found
Python 3 is not installed on your machine. Download and install it from the official site:
- python.org/downloads
- Download the latest Python 3 installer for your operating system.
- Run the installer. On Windows, check the box that says Add python.exe to PATH before clicking Install Now.
- Close and reopen your terminal, then run python3 --version again to confirm.
ⓧ I see Python 2.x
Your system is using Python 2, which is end-of-life. You need Python 3.
- Download and install the latest Python 3 from python.org/downloads.
- After installing, run python3 --version again to confirm you see a version starting with 3.
- Create a virtual environment in your project folder by running this command:
python3 -m venv venv- Activate the virtual environment by running:
macOS
source venv/bin/activateWindows
venv\Scripts\activateYou should see (venv) appear at the beginning of your terminal prompt.
- Install the psycopg3 database adapter by running:
pip install "psycopg[binary]"Why psycopg[binary]?
The [binary] extra bundles a pre-compiled C library so you do not need to compile anything locally. It gives you full-speed database communication out of the box.
You should see output ending with Successfully installed psycopg... in your terminal.
Seeing a pip error?
Make sure your virtual environment is active (you see (venv) in your prompt). If pip is not found, try python3 -m pip install "psycopg[binary]" instead.
Help me fix my pip installation
Write the schema setup script
This script connects to your fleet_tracking database, enables the required extensions, and creates a hypertable designed for GPS telemetry data.
- Create a new file called setup_schema.py in your project folder (the same folder that contains docker-compose.yml).
- Add the following code to connect to the database and create the hypertable:
import psycopg
DB_URL = "postgresql://postgres:password@localhost:5432/fleet_tracking"
def setup():
with psycopg.connect(DB_URL) as conn:
with conn.cursor() as cur:
# Enable extensions
cur.execute("CREATE EXTENSION IF NOT EXISTS timescaledb CASCADE;")
cur.execute("CREATE EXTENSION IF NOT EXISTS postgis;")
# Create the hypertable
cur.execute("""
CREATE TABLE IF NOT EXISTS vehicle_positions (
time TIMESTAMPTZ NOT NULL,
vehicle_id INTEGER NOT NULL,
latitude DOUBLE PRECISION NOT NULL,
longitude DOUBLE PRECISION NOT NULL,
speed DOUBLE PRECISION NOT NULL,
heading DOUBLE PRECISION NOT NULL,
geom geometry(Point, 4326)
) WITH (
tsdb.hypertable,
tsdb.segmentby = 'vehicle_id',
tsdb.orderby = 'time DESC',
tsdb.chunk_interval = '1 hour'
);
""")What does this code do?
The DB_URL connects to the fleet_tracking database running in your Docker container using the password you set in docker-compose.yml.
The two CREATE EXTENSION statements enable TimescaleDB (time-series partitioning) and PostGIS (geospatial queries) inside the database.
The CREATE TABLE ... WITH (tsdb.hypertable) statement does two things at once: it creates the table and immediately converts it into a hypertable. The tsdb.segmentby = 'vehicle_id' groups data by vehicle for efficient per-vehicle lookups. The tsdb.chunk_interval = '1 hour' creates a new partition every hour so old data can be compressed independently.
The geom column uses PostGIS geometry(Point, 4326) type, which stores GPS coordinates in the standard WGS84 coordinate system used by GPS devices worldwide.
- Still inside the with conn.cursor() as cur: block, add the following code below the hypertable creation to create the indexes:
# Create spatial index
cur.execute("""
CREATE INDEX IF NOT EXISTS idx_vehicle_positions_geom
ON vehicle_positions USING GIST (geom);
""")
# Create composite index for time-series lookups
cur.execute("""
CREATE INDEX IF NOT EXISTS idx_vehicle_positions_vid_time
ON vehicle_positions (vehicle_id, time DESC);
""")
conn.commit()
print("Schema setup complete.")What do these indexes do?
The GiST index on the geom column accelerates spatial queries like "find all vehicles within 2km of this point." Without it, the database would scan every row to calculate distances.
The composite index on (vehicle_id, time DESC) speeds up per-vehicle time-series lookups. When you query "show me vehicle 42's last 10 positions," the database jumps directly to that vehicle's most recent records.
- Below the print("Schema setup complete.") line (still inside the setup() function), add the verification logic and main guard:
# Verify hypertable creation
with psycopg.connect(DB_URL) as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT hypertable_name, num_chunks
FROM timescaledb_information.hypertables
WHERE hypertable_name = 'vehicle_positions';
""")
row = cur.fetchone()
if row:
print(f"Hypertable '{row[0]}' created successfully (chunks: {row[1]})")
else:
print("ERROR: Hypertable was not created.")
if __name__ == "__main__":
setup()What does the verification do?
This opens a fresh connection and queries the timescaledb_information.hypertables system view. This view lists all hypertables in the database, confirming that TimescaleDB correctly converted your table into a partitioned hypertable.
- Save setup_schema.py.
✔️ Awesome, I've got everything!
Great. Double check you have saved setup_schema.py before running it in the next substep.
ⓧ I'd like to double check the full code
Here is the complete setup_schema.py file. Compare your code against this:
import psycopg
DB_URL = "postgresql://postgres:password@localhost:5432/fleet_tracking"
def setup():
with psycopg.connect(DB_URL) as conn:
with conn.cursor() as cur:
# Enable extensions
cur.execute("CREATE EXTENSION IF NOT EXISTS timescaledb CASCADE;")
cur.execute("CREATE EXTENSION IF NOT EXISTS postgis;")
# Create the hypertable
cur.execute("""
CREATE TABLE IF NOT EXISTS vehicle_positions (
time TIMESTAMPTZ NOT NULL,
vehicle_id INTEGER NOT NULL,
latitude DOUBLE PRECISION NOT NULL,
longitude DOUBLE PRECISION NOT NULL,
speed DOUBLE PRECISION NOT NULL,
heading DOUBLE PRECISION NOT NULL,
geom geometry(Point, 4326)
) WITH (
tsdb.hypertable,
tsdb.segmentby = 'vehicle_id',
tsdb.orderby = 'time DESC',
tsdb.chunk_interval = '1 hour'
);
""")
# Create spatial index
cur.execute("""
CREATE INDEX IF NOT EXISTS idx_vehicle_positions_geom
ON vehicle_positions USING GIST (geom);
""")
# Create composite index for time-series lookups
cur.execute("""
CREATE INDEX IF NOT EXISTS idx_vehicle_positions_vid_time
ON vehicle_positions (vehicle_id, time DESC);
""")
conn.commit()
print("Schema setup complete.")
# Verify hypertable creation
with psycopg.connect(DB_URL) as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT hypertable_name, num_chunks
FROM timescaledb_information.hypertables
WHERE hypertable_name = 'vehicle_positions';
""")
row = cur.fetchone()
if row:
print(f"Hypertable '{row[0]}' created successfully (chunks: {row[1]})")
else:
print("ERROR: Hypertable was not created.")
if __name__ == "__main__":
setup()Run the setup script and verify the hypertable
Time to execute the script against your running TimescaleDB container.
- Make sure your Docker container is still running. You can confirm by running:
docker psYou should see the tsdb-geo container listed with status Up.
- Run the schema setup script by executing:
python3 setup_schema.pyYou should see output like this:
Schema setup complete.
Hypertable 'vehicle_positions' created successfully (chunks: 0)Why does it say 0 chunks?
Chunks are created on demand when data arrives. The hypertable starts empty, so there are zero chunks. Once you ingest data in the next step, TimescaleDB will automatically create hourly chunks based on the tsdb.chunk_interval = '1 hour' setting.
Seeing a connection refused error?
This means the database container is not running or not ready yet. Run docker ps to check that the tsdb-geo container shows healthy in the STATUS column. If it is still starting, wait 10 seconds and try again.
If the container is not listed at all, return to your project folder and run docker compose up -d to start it.
Help me debug my connection error
Your schema is locked in. Next up, you will build the ingestion script that generates 200,000 rows of GPS telemetry and streams them into this hypertable at high speed.
Build the High-Throughput Ingestion Script
Your hypertable is ready and waiting for data. Now you need a way to fill it with realistic GPS telemetry at high speed.
In this step, you will write a Python script that simulates 500 vehicles driving around San Francisco, then streams their positions into the database using the PostgreSQL COPY protocol. This protocol bypasses per-row INSERT overhead and loads data in bulk, achieving throughput orders of magnitude faster than individual inserts.
In this step, get ready to:
- Write data generator functions that simulate 500 vehicles with random-walk movement.
- Build an ingestion function using psycopg's COPY protocol to stream batches of 10,000 rows.
- Run the ingestion for 200,000 rows and measure throughput in rows per second.
Write the data generator functions
Before you can test ingestion performance, you need realistic data. The generator will initialize 500 vehicles at random positions near downtown San Francisco, then move each one using a random-walk pattern that adjusts speed and heading each tick.
- Create a new file called ingest.py in the same folder as setup_schema.py.
- Add the imports, configuration constants, and the generate_vehicles() function by pasting the following code:
import psycopg
import random
import time
from datetime import datetime, timezone, timedelta
DB_URL = "postgresql://postgres:password@localhost:5432/fleet_tracking"
NUM_VEHICLES = 500
BATCH_SIZE = 10000
TOTAL_ROWS = 200000
# City center (San Francisco)
CENTER_LAT = 37.7749
CENTER_LON = -122.4194
def generate_vehicles():
"""Initialize vehicle positions around the city center."""
vehicles = []
for i in range(NUM_VEHICLES):
lat = CENTER_LAT + random.uniform(-0.05, 0.05)
lon = CENTER_LON + random.uniform(-0.05, 0.05)
speed = random.uniform(0, 60)
heading = random.uniform(0, 360)
vehicles.append({
"id": i,
"lat": lat,
"lon": lon,
"speed": speed,
"heading": heading,
})
return vehiclesWhat does this code do?
The constants at the top define the simulation parameters: 500 vehicles, batches of 10,000 rows, and a total target of 200,000 rows.
The generate_vehicles() function creates a list of 500 vehicle dictionaries. Each vehicle starts at a random position within 0.05 degrees of downtown San Francisco (latitude 37.7749, longitude -122.4194), with a random initial speed and heading.
- Below generate_vehicles(), add the move_vehicle() function that simulates realistic movement:
def move_vehicle(vehicle):
"""Simulate vehicle movement with random walk."""
import math
# Slightly adjust heading
vehicle["heading"] += random.uniform(-15, 15)
vehicle["heading"] %= 360
# Adjust speed
vehicle["speed"] = max(0, min(80, vehicle["speed"] + random.uniform(-5, 5)))
# Move based on speed and heading (approximate degrees per second at this latitude)
distance_deg = (vehicle["speed"] / 3600) * 0.00001 # rough conversion
heading_rad = math.radians(vehicle["heading"])
vehicle["lat"] += distance_deg * math.cos(heading_rad)
vehicle["lon"] += distance_deg * math.sin(heading_rad)
# Keep within bounds
vehicle["lat"] = max(CENTER_LAT - 0.1, min(CENTER_LAT + 0.1, vehicle["lat"]))
vehicle["lon"] = max(CENTER_LON - 0.1, min(CENTER_LON + 0.1, vehicle["lon"]))What does this code do?
Each call to move_vehicle() nudges the vehicle's heading by up to 15 degrees in either direction and adjusts speed within a 0-80 km/h range. It then converts speed and heading into a latitude/longitude displacement using basic trigonometry.
The bounding box keeps all vehicles within 0.1 degrees of the city center, preventing them from drifting off into the ocean.
- Save ingest.py.
Seeing a syntax error?
Make sure the move_vehicle() function is at the same indentation level as generate_vehicles() (not nested inside it). Both should be top-level function definitions.
help me fix a syntax error in my ingest.py vehicle generator functions
Write the COPY-based ingestion function
The COPY protocol streams rows directly into PostgreSQL in a single transaction with minimal per-row overhead. psycopg exposes this via cursor.copy(), which returns a context manager you can feed rows into using write_row().
- Below move_vehicle(), add the generate_batch() function that produces a batch of position records:
def generate_batch(vehicles, base_time, batch_size):
"""Generate a batch of position records."""
rows = []
records_per_vehicle = batch_size // NUM_VEHICLES
current_time = base_time
for _ in range(records_per_vehicle):
current_time += timedelta(seconds=1)
for vehicle in vehicles:
move_vehicle(vehicle)
wkt = f"SRID=4326;POINT({vehicle['lon']} {vehicle['lat']})"
rows.append((
current_time,
vehicle["id"],
vehicle["lat"],
vehicle["lon"],
vehicle["speed"],
vehicle["heading"],
wkt,
))
return rows, current_timeWhat does this code do?
Each batch produces 10,000 rows (20 time ticks multiplied by 500 vehicles). For each tick, it advances the clock by one second, moves every vehicle, and builds a tuple containing all the column values.
The wkt variable holds a Well-Known Text geometry string (SRID=4326;POINT(lon lat)) that PostGIS can parse directly during the COPY operation.
- Below generate_batch(), add the ingest() function and the __main__ block:
def ingest():
vehicles = generate_vehicles()
base_time = datetime.now(timezone.utc) - timedelta(hours=1)
total_ingested = 0
start = time.time()
with psycopg.connect(DB_URL) as conn:
while total_ingested < TOTAL_ROWS:
batch, base_time = generate_batch(vehicles, base_time, BATCH_SIZE)
with conn.cursor() as cur:
with cur.copy(
"COPY vehicle_positions (time, vehicle_id, latitude, longitude, speed, heading, geom) FROM STDIN"
) as copy:
for row in batch:
copy.write_row(row)
conn.commit()
total_ingested += len(batch)
elapsed = time.time() - start
rate = total_ingested / elapsed
print(f"Ingested {total_ingested:,} rows | {rate:,.0f} rows/sec")
elapsed = time.time() - start
print(f"\nDone! {total_ingested:,} rows in {elapsed:.1f}s ({total_ingested/elapsed:,.0f} rows/sec)")
if __name__ == "__main__":
ingest()What does this code do?
The ingest() function opens a single database connection and loops until 200,000 rows are ingested. Each iteration generates a batch of 10,000 rows, then streams them into the database using cur.copy().
The COPY vehicle_positions FROM STDIN statement tells PostgreSQL to accept streamed row data. write_row() feeds each tuple into the stream, and psycopg handles type adaptation automatically.
After each batch, the transaction commits and the script prints a running tally of total rows and current throughput.
- Save ingest.py.
Why COPY instead of INSERT?
A standard INSERT statement processes one row at a time, parsing the SQL and checking constraints for each. COPY uses a dedicated binary protocol that streams all rows in a single operation with minimal per-row overhead.
For bulk loading into TimescaleDB, COPY is the recommended approach because it achieves the highest throughput on a single connection.
Seeing a connection error?
Make sure your TimescaleDB container is still running. Run docker ps and confirm you see tsdb-geo in the list. If it's not there, start it again with docker compose up -d.
help me fix a database connection error in my ingest.py script
Run the ingestion and measure throughput
The script includes timing instrumentation that reports rows per second after each batch. You will see throughput numbers climb as the ingestion runs through all 200,000 rows.
- Run the ingestion script by executing this command:
python ingest.pyYou should see output like this, with the row count and throughput updating after each batch:
Ingested 10,000 rows | 45,000 rows/sec
Ingested 20,000 rows | 52,000 rows/sec
...
Ingested 200,000 rows | 80,000 rows/sec
Done! 200,000 rows in 2.8s (71,428 rows/sec)What should I see?
The exact throughput depends on your machine. On most systems you should see somewhere between 50,000 and 150,000+ rows per second. The script prints a final summary line when it finishes all 200,000 rows.
Seeing very low throughput or timeouts?
If throughput is below 10,000 rows/sec, check that Docker Desktop has at least 4 GB of memory allocated. On macOS or Windows, open Docker Desktop settings and increase the memory limit under **Resources**.
help me improve my TimescaleDB COPY ingestion throughput
✔️ Awesome, I've got everything!
Great. Double check you have saved ingest.py and the ingestion completed successfully.
ⓧ I'd like to double check the full code
Here is the complete ingest.py file:
import psycopg
import random
import time
from datetime import datetime, timezone, timedelta
DB_URL = "postgresql://postgres:password@localhost:5432/fleet_tracking"
NUM_VEHICLES = 500
BATCH_SIZE = 10000
TOTAL_ROWS = 200000
# City center (San Francisco)
CENTER_LAT = 37.7749
CENTER_LON = -122.4194
def generate_vehicles():
"""Initialize vehicle positions around the city center."""
vehicles = []
for i in range(NUM_VEHICLES):
lat = CENTER_LAT + random.uniform(-0.05, 0.05)
lon = CENTER_LON + random.uniform(-0.05, 0.05)
speed = random.uniform(0, 60)
heading = random.uniform(0, 360)
vehicles.append({
"id": i,
"lat": lat,
"lon": lon,
"speed": speed,
"heading": heading,
})
return vehicles
def move_vehicle(vehicle):
"""Simulate vehicle movement with random walk."""
import math
# Slightly adjust heading
vehicle["heading"] += random.uniform(-15, 15)
vehicle["heading"] %= 360
# Adjust speed
vehicle["speed"] = max(0, min(80, vehicle["speed"] + random.uniform(-5, 5)))
# Move based on speed and heading (approximate degrees per second at this latitude)
distance_deg = (vehicle["speed"] / 3600) * 0.00001 # rough conversion
heading_rad = math.radians(vehicle["heading"])
vehicle["lat"] += distance_deg * math.cos(heading_rad)
vehicle["lon"] += distance_deg * math.sin(heading_rad)
# Keep within bounds
vehicle["lat"] = max(CENTER_LAT - 0.1, min(CENTER_LAT + 0.1, vehicle["lat"]))
vehicle["lon"] = max(CENTER_LON - 0.1, min(CENTER_LON + 0.1, vehicle["lon"]))
def generate_batch(vehicles, base_time, batch_size):
"""Generate a batch of position records."""
rows = []
records_per_vehicle = batch_size // NUM_VEHICLES
current_time = base_time
for _ in range(records_per_vehicle):
current_time += timedelta(seconds=1)
for vehicle in vehicles:
move_vehicle(vehicle)
wkt = f"SRID=4326;POINT({vehicle['lon']} {vehicle['lat']})"
rows.append((
current_time,
vehicle["id"],
vehicle["lat"],
vehicle["lon"],
vehicle["speed"],
vehicle["heading"],
wkt,
))
return rows, current_time
def ingest():
vehicles = generate_vehicles()
base_time = datetime.now(timezone.utc) - timedelta(hours=1)
total_ingested = 0
start = time.time()
with psycopg.connect(DB_URL) as conn:
while total_ingested < TOTAL_ROWS:
batch, base_time = generate_batch(vehicles, base_time, BATCH_SIZE)
with conn.cursor() as cur:
with cur.copy(
"COPY vehicle_positions (time, vehicle_id, latitude, longitude, speed, heading, geom) FROM STDIN"
) as copy:
for row in batch:
copy.write_row(row)
conn.commit()
total_ingested += len(batch)
elapsed = time.time() - start
rate = total_ingested / elapsed
print(f"Ingested {total_ingested:,} rows | {rate:,.0f} rows/sec")
elapsed = time.time() - start
print(f"\nDone! {total_ingested:,} rows in {elapsed:.1f}s ({total_ingested/elapsed:,.0f} rows/sec)")
if __name__ == "__main__":
ingest()Your database now holds 200,000 GPS telemetry records partitioned across time-based chunks. Next up, you will add geospatial indexing and write spatial-temporal queries that combine location filtering with time-series aggregations.
Run Spatial-Temporal Queries
You have 200,000 GPS telemetry records sitting in your hypertable. Now comes the real payoff: querying that data by both time and location simultaneously.
Spatial-temporal queries answer questions like "which vehicles were near this intersection in the last 5 minutes?" This is the core pattern behind fleet management dashboards, delivery ETAs, and ride-hailing dispatch systems.
In this step, get ready to:
- Write a proximity query that finds vehicles within 2km of a point using PostGIS.
- Write a time_bucket aggregation query that computes average speed per minute.
- Run the queries and check current storage size.
Write the spatial-temporal proximity query
The most powerful feature of combining TimescaleDB with PostGIS is filtering by both distance and time in a single query. The ST_DWithin function checks whether two geometries are within a given distance of each other. When you cast to the geography type, the distance parameter is in meters.
- Create a new file called queries.py in your project folder and add the imports, connection string, and spatial query function:
import psycopg
from datetime import datetime, timezone, timedelta
DB_URL = "postgresql://postgres:password@localhost:5432/fleet_tracking"
def spatial_temporal_query():
"""Find vehicles within 2km of a point in the last 5 minutes."""
print("\n--- Spatial-Temporal Query ---")
print("Finding vehicles within 2km of downtown SF in the last 5 minutes...\n")
with psycopg.connect(DB_URL) as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT
vehicle_id,
time,
latitude,
longitude,
speed
FROM vehicle_positions
WHERE time > now() - INTERVAL '5 minutes'
AND ST_DWithin(
geom::geography,
ST_SetSRID(ST_MakePoint(-122.4194, 37.7749), 4326)::geography,
2000
)
ORDER BY time DESC
LIMIT 20;
""")
rows = cur.fetchall()
print(f"Found {len(rows)} records (showing up to 20):")
for row in rows:
print(f" Vehicle {row[0]:3d} | {row[1]} | ({row[2]:.4f}, {row[3]:.4f}) | {row[4]:.1f} km/h")What does this query do?
The WHERE clause combines two filters. First, time > now() - INTERVAL '5 minutes' restricts to recent data. TimescaleDB's chunk pruning skips entire partitions outside this window.
Second, ST_DWithin(geom::geography, ..., 2000) filters to records within 2000 meters of downtown San Francisco. Casting to geography ensures the distance is calculated in meters on a sphere, not degrees on a flat plane.
The GiST index you created earlier makes this spatial filter fast. Without it, PostgreSQL would scan every row.
- Save queries.py.
Add the time_bucket aggregation query
TimescaleDB's time_bucket function groups timestamps into fixed intervals. This lets you compute metrics like average speed per minute without writing complex date math.
- In queries.py, add the time_bucket_query() function below spatial_temporal_query():
def time_bucket_query():
"""Compute average speed per vehicle per 1-minute bucket."""
print("\n--- Time Bucket Aggregation ---")
print("Average speed per vehicle per 1-minute bucket (top 10)...\n")
with psycopg.connect(DB_URL) as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT
time_bucket('1 minute', time) AS bucket,
vehicle_id,
AVG(speed) AS avg_speed,
COUNT(*) AS num_readings
FROM vehicle_positions
WHERE time > now() - INTERVAL '10 minutes'
GROUP BY bucket, vehicle_id
ORDER BY bucket DESC, avg_speed DESC
LIMIT 10;
""")
rows = cur.fetchall()
print(f"{'Bucket':<25} {'Vehicle':<10} {'Avg Speed':<12} {'Readings'}")
print("-" * 60)
for row in rows:
print(f"{str(row[0]):<25} {row[1]:<10} {row[2]:>8.1f} km/h {row[3]:>5}")What does this query do?
The time_bucket('1 minute', time) expression rounds each timestamp down to the nearest minute boundary. This creates a bucket column you can GROUP BY.
The query then computes AVG(speed) and COUNT(*) per vehicle per bucket. This is the foundation of any time-series dashboard, whether you are tracking fleet speeds, sensor temperatures, or server response times.
- Save queries.py.
Run the queries and check storage size
Before you run the queries, you will add one more function that reports the current storage footprint of your hypertable. This gives you a baseline to compare against after compression in the next step.
- Add the compression_stats() function and the __main__ block at the bottom of queries.py:
def compression_stats():
"""Show compression statistics."""
print("\n--- Compression Stats ---")
with psycopg.connect(DB_URL) as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT
pg_size_pretty(hypertable_size('vehicle_positions')) AS total_size,
pg_size_pretty(pg_total_relation_size('vehicle_positions')) AS relation_size;
""")
row = cur.fetchone()
print(f"Hypertable total size: {row[0]}")
print(f"Relation size: {row[1]}")
if __name__ == "__main__":
spatial_temporal_query()
time_bucket_query()
compression_stats()What does this code do?
The hypertable_size() function returns the total disk space used by the hypertable including indexes and toast data. The pg_total_relation_size() function shows the raw relation size for comparison.
The __main__ block runs all three queries in sequence when you execute the file.
- Save queries.py.
- Run the queries script by running this command:
python queries.pyYou should see three sections of output. The spatial-temporal query prints vehicle records found near downtown SF. The time bucket query prints a table of average speeds grouped by minute. The compression stats section prints the current hypertable size (likely 40-80 MB depending on your system).
Seeing zero results from the spatial query?
The spatial query filters to the last 5 minutes. If you ran the ingestion script more than 5 minutes ago, the most recent data may have fallen outside that window. Re-run python ingest.py to load fresh data, then run python queries.py again immediately.
If you see a connection error, make sure your TimescaleDB container is still running with docker ps.
Help me debug why my spatial-temporal query returns no results.
✔️ Awesome, I've got everything!
Great. Make sure you have saved queries.py with all three functions.
ⓧ I'd like to double check the full code
import psycopg
from datetime import datetime, timezone, timedelta
DB_URL = "postgresql://postgres:password@localhost:5432/fleet_tracking"
def spatial_temporal_query():
"""Find vehicles within 2km of a point in the last 5 minutes."""
print("\n--- Spatial-Temporal Query ---")
print("Finding vehicles within 2km of downtown SF in the last 5 minutes...\n")
with psycopg.connect(DB_URL) as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT
vehicle_id,
time,
latitude,
longitude,
speed
FROM vehicle_positions
WHERE time > now() - INTERVAL '5 minutes'
AND ST_DWithin(
geom::geography,
ST_SetSRID(ST_MakePoint(-122.4194, 37.7749), 4326)::geography,
2000
)
ORDER BY time DESC
LIMIT 20;
""")
rows = cur.fetchall()
print(f"Found {len(rows)} records (showing up to 20):")
for row in rows:
print(f" Vehicle {row[0]:3d} | {row[1]} | ({row[2]:.4f}, {row[3]:.4f}) | {row[4]:.1f} km/h")
def time_bucket_query():
"""Compute average speed per vehicle per 1-minute bucket."""
print("\n--- Time Bucket Aggregation ---")
print("Average speed per vehicle per 1-minute bucket (top 10)...\n")
with psycopg.connect(DB_URL) as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT
time_bucket('1 minute', time) AS bucket,
vehicle_id,
AVG(speed) AS avg_speed,
COUNT(*) AS num_readings
FROM vehicle_positions
WHERE time > now() - INTERVAL '10 minutes'
GROUP BY bucket, vehicle_id
ORDER BY bucket DESC, avg_speed DESC
LIMIT 10;
""")
rows = cur.fetchall()
print(f"{'Bucket':<25} {'Vehicle':<10} {'Avg Speed':<12} {'Readings'}")
print("-" * 60)
for row in rows:
print(f"{str(row[0]):<25} {row[1]:<10} {row[2]:>8.1f} km/h {row[3]:>5}")
def compression_stats():
"""Show compression statistics."""
print("\n--- Compression Stats ---")
with psycopg.connect(DB_URL) as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT
pg_size_pretty(hypertable_size('vehicle_positions')) AS total_size,
pg_size_pretty(pg_total_relation_size('vehicle_positions')) AS relation_size;
""")
row = cur.fetchone()
print(f"Hypertable total size: {row[0]}")
print(f"Relation size: {row[1]}")
if __name__ == "__main__":
spatial_temporal_query()
time_bucket_query()
compression_stats()Your queries prove that time-series partitioning and geospatial indexing work together seamlessly. Next up, you will enable compression to dramatically shrink the storage footprint of your historical data.
Enable Compression and Measure Storage Savings
Your hypertable has 200,000 rows of geospatial telemetry data, and your spatial-temporal queries are returning results. In production, fleet tracking data grows fast. A single fleet of 500 vehicles reporting every second generates over 43 million rows per day.
TimescaleDB's columnstore compression can reduce storage by up to 98%. In this step, you'll compress your existing data chunks, measure the storage savings, and confirm that your spatial queries still work correctly on compressed data.
In this step, get ready to:
- Compress all existing chunks in your hypertable and measure the before/after storage sizes.
- Verify that spatial-temporal queries still return correct results on compressed data.
- Observe the dramatic storage reduction that makes long-term retention practical.
Create the compression script
When you created the hypertable in setup_schema.py, the tsdb.segmentby and tsdb.orderby options defined how data gets organized during compression. Now you need a script that triggers the actual compression on existing chunks.
- Create a new file called compress.py in the same folder as your other scripts, then add the following code:
import psycopg
DB_URL = "postgresql://postgres:password@localhost:5432/fleet_tracking"
def enable_compression():
"""Enable and run compression on the hypertable."""
print("Checking hypertable size before compression...")
with psycopg.connect(DB_URL) as conn:
with conn.cursor() as cur:
# Get size before compression
cur.execute("""
SELECT pg_size_pretty(hypertable_size('vehicle_positions'));
""")
before_size = cur.fetchone()[0]
print(f"Size before compression: {before_size}")
# Manually compress all chunks
print("\nCompressing all chunks...")
cur.execute("""
SELECT compress_chunk(c)
FROM show_chunks('vehicle_positions') c;
""")
compressed_count = cur.rowcount
print(f"Compressed {compressed_count} chunks.")
conn.commit()
# Check size after
with psycopg.connect(DB_URL) as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT pg_size_pretty(hypertable_size('vehicle_positions'));
""")
after_size = cur.fetchone()[0]
print(f"Size after compression: {after_size}")
print(f"\nCompression complete! ({before_size} -> {after_size})")
if __name__ == "__main__":
enable_compression()What does this code do?
This script measures the hypertable size before and after compression so you can see exactly how much space you save.
- hypertable_size('vehicle_positions') returns the total disk space used by the hypertable, including indexes and toast data.
- compress_chunk(c) FROM show_chunks('vehicle_positions') iterates over every chunk in the hypertable and compresses it into columnar format.
- The script opens a fresh connection after compression to get an accurate post-compression size reading.
Why manual compression instead of a policy?
In production, you would use add_compression_policy() to automatically compress chunks older than a threshold. For this project, manual compression lets you see the immediate before/after effect in a single script run.
- Save compress.py.
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before moving on.
ⓧ I'd like to double check the full code
import psycopg
DB_URL = "postgresql://postgres:password@localhost:5432/fleet_tracking"
def enable_compression():
"""Enable and run compression on the hypertable."""
print("Checking hypertable size before compression...")
with psycopg.connect(DB_URL) as conn:
with conn.cursor() as cur:
# Get size before compression
cur.execute("""
SELECT pg_size_pretty(hypertable_size('vehicle_positions'));
""")
before_size = cur.fetchone()[0]
print(f"Size before compression: {before_size}")
# Manually compress all chunks
print("\nCompressing all chunks...")
cur.execute("""
SELECT compress_chunk(c)
FROM show_chunks('vehicle_positions') c;
""")
compressed_count = cur.rowcount
print(f"Compressed {compressed_count} chunks.")
conn.commit()
# Check size after
with psycopg.connect(DB_URL) as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT pg_size_pretty(hypertable_size('vehicle_positions'));
""")
after_size = cur.fetchone()[0]
print(f"Size after compression: {after_size}")
print(f"\nCompression complete! ({before_size} -> {after_size})")
if __name__ == "__main__":
enable_compression()Run compression and measure storage savings
With 200,000 rows of GPS telemetry sitting uncompressed in your hypertable, this is a good moment to see what columnar compression does to the storage footprint.
- Run the compression script by executing this command:
python compress.pyYou should see output showing the before and after sizes, plus the number of chunks compressed. The exact sizes depend on your system, but you should see a significant reduction (often 80-90%+ for this type of repetitive numeric data).
What should I see?
Your output will look something like this (exact values vary):
Checking hypertable size before compression...
Size before compression: 40 MB
Compressing all chunks...
Compressed 2 chunks.
Size after compression: 5608 kB
Compression complete! (40 MB -> 5608 kB)The dramatic size drop happens because TimescaleDB converts the row-oriented data into column-oriented format. Columns of similar values (like vehicle_id repeating 0-499, or latitude values clustered around 37.77) compress extremely well.
Seeing an error about compression not being enabled?
If you get an error mentioning that the hypertable doesn't support compression, make sure your setup_schema.py used the tsdb.segmentby and tsdb.orderby options when creating the table. These options configure the compression settings.
help me fix a compression error in TimescaleDB
Verify queries work on compressed data
Compression changes how data is stored on disk, but TimescaleDB transparently decompresses data at query time. Your existing queries should return the same results without any changes.
- Run the queries script to confirm the spatial-temporal query still works on compressed data:
python queries.pyYou should see the same spatial-temporal query results, time bucket aggregations, and compression stats as before. The key difference is that the compression stats now show the reduced size.
Why does this matter?
Transparent query support on compressed data means you get massive storage savings without changing any application code. Your ST_DWithin queries and time_bucket aggregations work identically whether the underlying chunks are compressed or not.
In production fleet systems, this is critical. You can retain months of historical GPS data at a fraction of the storage cost while still running analytics over the full dataset.
Query returning zero results?
The spatial-temporal query filters by time > now() - INTERVAL '5 minutes'. If too much time has passed since you ran ingest.py, the most recent data may be older than 5 minutes. Re-run ingest.py to add fresh data, then try queries.py again.
help me debug why my spatial query returns no results after compression
Your pipeline now handles the full lifecycle: high-throughput ingestion, spatial-temporal queries, and efficient long-term storage via compression. Next up, the Secret Mission challenges you to build a continuous aggregate that pre-computes vehicle density heatmaps using H3 hexagonal grids.
Secret mission
Continuous Aggregate with H3 Hexagonal Density
Your pipeline can ingest and query geospatial time-series data at high speed. But what if you need a pre-computed heatmap showing where vehicles concentrate over time? In this secret mission, you'll build a continuous aggregate that materializes vehicle density into H3 hexagonal grid cells every 5 minutes.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally via Docker, so there are no ongoing cloud costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- Docker container tsdb-geo running TimescaleDB with PostGIS.
- Docker volume tsdb_data storing the database data.
- Python virtual environment with psycopg installed.
- Project files: docker-compose.yml, setup_schema.py, ingest.py, queries.py, compress.py, and h3_density.py.
Keep everything running
No action needed. Choose this if you are still actively building or experimenting with your pipeline.
- Your tsdb-geo container continues to run and uses memory on your machine.
- The database data persists in the tsdb_data Docker volume.
- There is no financial cost. Everything runs locally.
Pause - I'll come back to this later
Stop the running container to free up memory while keeping all your data intact for next time.
- Open your terminal and navigate to the folder containing your docker-compose.yml file.
- Stop the container by running:
docker compose stop- This stops the container without removing it or its data.
- When you want to resume, run docker compose start from the same folder.
Delete - I don't want to use this again
Remove all project resources and start fresh. This deletes your container, database volume, virtual environment, and project files.
- Open your terminal and navigate to the folder containing your docker-compose.yml file.
- Stop and remove the container and its volume by running:
docker compose down -v- The -v flag removes the named tsdb_data volume along with the container.
- Deactivate your virtual environment (if active) by running:
deactivate- Navigate one level up from your project folder, then delete it.
- On macOS/Linux, run:
cd ~/Desktop
rm -rf fleet-tracking- On Windows, run:
cd ~\Desktop
rmdir /s /q fleet-tracking- This removes all Python scripts, the virtual environment, and the docker-compose.yml file.
Nice Work!
Nice Work!
Nice work! You just built a high-throughput geospatial time-series ingestion pipeline that streams 200,000 GPS telemetry records into TimescaleDB at over 100,000 rows per second.
You've learned how to:
- Implement a Python ingestion script using the PostgreSQL COPY protocol to achieve high-throughput bulk loading into a hypertable.
- Design a hypertable schema with automatic time-based partitioning, GiST spatial indexing, and PostGIS geometry columns for combined time-series and location workloads.
- Write spatial-temporal queries combining ST_DWithin distance filtering with time_bucket aggregations, and enable automatic compression for significant storage savings on historical data.
- Created a continuous aggregate with H3 hexagonal grid cells for pre-computed vehicle density heatmaps.
Ready to quiz yourself?