Horizontally Scaling PostgreSQL: Database Sharding with Citus

A practical guide to scaling your database infrastructure horizontally. Learn how to shard your data, optimize query routing, and build a distributed multi-tenant architecture using Postgres, Citus and Docker.

Prerequisites

What do you need to follow along?

1. A working Laptop/PC?
2. Working knowledge of SQL or PostgreSQL.
3. To have at least eaten something today (preferably swallow related).

Database sharding is a strategy used to solve scalability issues in applications that have a massive amount of data. It involves splitting data into smaller chunks called shards. Each shard is stored on a separate database server instance, effectively spreading the data and the workload across multiple machines.

Why Shard Data?

As an application grows, the number of users and the amount of data it stores increase over time. At a certain limit, a single database server will become a bottleneck. The application slows down and affects the user experience.

Instead of upgrading a single server with more CPU or RAM, which will eventually hit a hardware ceiling, sharding lets you add more servers to your cluster. This allows applications to handle nearly infinite growth in data volume and user traffic.

Understanding Shards and Shard Keys

Before we start sharding data, let's first understand what shards and shard keys are.

Shards

The partitioned data chunks are called logical shards. The machine that stores the logical shard is called a physical shard or database node. A physical shard can contain multiple logical shards.

Shard Key

A shard key is used to determine how to partition the dataset. A shard key is a specific column in your dataset that determines which rows of data group together to form a shard.

Sharding Methods

There are a few ways to shard data depending on the application requirements.

1. Key-based sharding (hash sharding): Hash sharding uses a mathematical formula called a hash function. The hash function takes your shard key and generates a hash value, this hash value determines the exact physical shard where the row will be stored.
2. Range-based sharding: Range-based sharding, or dynamic sharding, splits database rows based on a range of values. Then the database designer assigns a shard key to the respective range.
3. Directory-based sharding: Directory sharding uses a lookup table to match database information to the corresponding physical shard. A lookup table is like a table on a spreadsheet that links a database column to a shard key.
4. Geo sharding: Geo sharding splits and stores database information according to geographical location.

For this guide, we are focusing exclusively on Hash Sharding. This is the industry standard for multi-tenant applications like the e-commerce platform we will be building in this tutorial.

Choosing Your Shard Key

If you pick the wrong shard key, your database will still crash under load. When choosing a shard key, you must ensure the data is distributed evenly across your nodes:

1. Cardinality: This refers to the number of unique values a key can have. You want a key with high cardinality like a User ID, not low cardinality like a Gender or State column, so the data can be split into many small, even chunks.
2. Frequency: You must avoid keys where one specific value appears too often. For example, if you shard by a City column and 80% of your users live in Ikeja, the Ikeja node will be overloaded while the others sit idle.
3. Monotonic change: This is the rate of change of the shard key. A monotonically increasing or decreasing shard key results in unbalanced shards. For example, consider a feedback database that is split into three different physical shards as follows:
   • Shard A stores feedback from customers who have made 0 to 10 purchases.
   • Shard B stores feedback from customers who have made 11 to 20 purchases.
   • Shard C stores feedback from customers who have made 21 or more purchases. As the business grows, customers will make 21 or more purchases. The application stores their feedback in Shard C. This results in an unbalanced shard because Shard C contains more feedback records than the other shards.

Is Sharding Actually For You?

While sharding solves massive scale requirements, it introduces significant technical and operational tradeoffs. You should not shard your database unless a single, heavily provisioned Postgres instance can no longer handle your traffic.

If you do decide to shard your database, here are some of the biggest challenges you will face:

1. Data hotspots: Even with a good hash function, uneven traffic can create hotspots where one server becomes overloaded by a small group of highly active users.
2. Performance regression: Queries that do not use the shard key require the system to scatter and gather across every server. Cross-shard joins are computationally expensive because the application must pull and merge data from multiple physical locations over the network.
3. Operational complexity: Managing a sharded environment increases the difficulty of routine tasks. Schema updates, backups, and point-in-time recovery must be coordinated across multiple independent instances.

How Can We Shard Our Postgres Database: Introducing Citus

Citus is an open-source extension that transforms standard PostgreSQL into a distributed database. Instead of being limited to a single server, Citus allows your database to scale horizontally across multiple machines.

How Citus Works

1. The Coordinator Node: When you ask your database a question, you talk to the coordinator. The coordinator directs your request to the right servers, gathers the answers, and combines them for your application.
2. Worker Nodes: Citus breaks down large tables into smaller, manageable chunks (shards) and spreads them across these worker servers. They do the heavy lifting of storing the data and running the queries.

Start Early, Scale Later

Migrating a traditional, monolithic PostgreSQL database to a distributed, sharded database later in its life is incredibly difficult. You have to change your schema, rewrite complex queries, and alter how your application connects to the database.

To avoid this nightmare, you can start using Citus on a single node from day one. You design your tables with distribution keys immediately, and your application code is written to query a distributed database from the start.

The Transition: When your data outgrows that single machine, you won't have to rewrite your application. You simply spin up new worker machines, and Citus moves the existing shards over to the new hardware in the background while the application keeps running.

Running Citus Locally

We will use Docker to run Citus locally. Why use Docker instead of adding the extension directly to your host machine's Postgres installation?

Simulating a distributed database means running multiple Postgres instances simultaneously. Doing this natively forces you to manually manage conflicting ports and multiple data directories. Docker Compose solves this by orchestrating the entire cluster's networking with a single command, giving you a clean state that can be torn down and rebuilt in seconds.

Single Node vs Multi-Node Setup

While it is possible to run a single node Citus instance using a simple docker run command:

docker run -d --name citus_standalone -p 5432:5432 -e POSTGRES_PASSWORD=postgres citusdata/citus:14.1.0

This guide will focus exclusively on setting up a Multi-Node Cluster. If you are interested in exploring the single node setup further, you can check out the official Citus documentation here.

For our distributed, multi-node cluster, we will use the following Docker Compose configuration:

services:
  master:
    container_name: "${PROJECT_NAME:-citus}_master"
    image: "citusdata/citus:14.1.0"
    ports: ["${COORDINATOR_EXTERNAL_PORT:-5433}:5432"]
    labels: ["com.citusdata.role=Master"]
    environment: &AUTH
      POSTGRES_USER: "${POSTGRES_USER:-postgres}"
      POSTGRES_PASSWORD: "${POSTGRES_PASSWORD}"
      PGUSER: "${POSTGRES_USER:-postgres}"
      PGPASSWORD: "${POSTGRES_PASSWORD}"
      POSTGRES_HOST_AUTH_METHOD: "${POSTGRES_HOST_AUTH_METHOD:-trust}"
  worker:
    image: "citusdata/citus:14.1.0"
    labels: ["com.citusdata.role=Worker"]
    environment: *AUTH

Citus has a built-in initialization script in its official Docker image. When launched via Docker Compose with these specific labels like com.citusdata.role, the coordinator automatically detects and registers the workers for you.

We can build and start the cluster in the background with:

docker compose up -d

We can also scale up dynamically. If you want to spin up a second or third worker machine on the fly, you do not even need to edit the compose file. Just run:

docker compose up -d --scale worker=3

Where 3 is the total number of workers you want to have running.

Simply spinning up the worker containers is not enough. After adding new worker nodes, we must register them with the coordinator so Citus knows they are active and ready to hold data. To do this, we will first connect to the interactive Postgres terminal on the master node:

docker compose exec master psql -U postgres

Now we can register our workers:

SELECT citus_add_node('db-sharding-worker-1', 5432);
SELECT citus_add_node('db-sharding-worker-2', 5432);
SELECT citus_add_node('db-sharding-worker-3', 5432);

So, if you scale up again to say 5 workers, you will have to run this command 2 more times to register the 2 new workers.

Notice the db-sharding- prefix in the hostnames. This is necessary because Docker Compose prefixes your project folder name to the container's network alias.

A Real World Scenario

In order to showcase the functionality of this database sharding setup, let's try to work with a real world scenario.

Consider an e-commerce multi-tenant application. It serves multiple independent companies that all share the same database infrastructure. To scale this effectively, we need to use a hash-based distribution with a well-chosen shard key. Let's assume by design, every tenant has a shared company_id, we will use that as our shard key.

By using company_id as the shard key across all operational tables, Citus guarantees that all data belonging to a specific company (like its users, its products, and its orders) is physically co-located on the exact same worker node.

Now that our workers are registered, we can create our core tables directly in the master node. Citus will automatically handle the underlying sharding for us.

Please note that the following schema is highly simplified to focus on the core concepts of sharding.

Run the following SQL commands in your coordinator's postgres interactive terminal:

-- 1. Create Tenant Table
CREATE TABLE companies (
    company_id INT PRIMARY KEY,
    company_name TEXT NOT NULL
);

-- 2. Create Tenant-Scoped Orders Table
CREATE TABLE orders (
    company_id INT NOT NULL,
    order_id INT NOT NULL,
    user_id INT NOT NULL,
    status_id INT NOT NULL, -- References our global order status table
    order_amount NUMERIC(10,2) NOT NULL,
    PRIMARY KEY (company_id, order_id)
);

-- 3. Create Global Order Reference Table
CREATE TABLE order_statuses (
    status_id INT PRIMARY KEY,
    status_name TEXT NOT NULL
);

-- 4. Create Tenant-Scoped Users Table
CREATE TABLE users (
    company_id INT NOT NULL,
    user_id INT NOT NULL,
    username TEXT NOT NULL,
    PRIMARY KEY (company_id, user_id)
);

-- 5. Create Tenant-Scoped Products Table
CREATE TABLE products (
    company_id INT NOT NULL,
    product_id INT NOT NULL,
    product_name TEXT NOT NULL,
    price NUMERIC(10,2) NOT NULL,
    PRIMARY KEY (company_id, product_id)
);

Telling Citus How to Shard and Replicate

Now that we have created our core tables, we have to tell Citus how to shard and replicate them across our worker nodes.

Reference Tables

First, we will set order_statuses as a reference table. We do this using create_reference_table.

Why do we create a reference table? A reference table is designed for relatively small, slowly-changing data like user roles, country codes, or order statuses.

When you mark a table as a reference table, Citus physically replicates a full copy of that entire table to every single worker node. This bypasses network overhead completely during JOIN operations. Because the worker node already has the data locally, it does not need to fetch it from another node to complete a query.

Distributed Tables

Next, we need to distribute our large tenant-data tables. We do this using create_distributed_table.

This function is the core of Citus's sharding mechanism. It tells the coordinator to partition the table based on a specific column (our shard key, company_id), and spread those partitions across the available worker nodes.

Because we use company_id across all these tables, any query filtering by a specific company will be routed directly to a single node, making it incredibly fast.

-- Replicate order_statuses to ALL nodes
SELECT create_reference_table('order_statuses');

-- Shard data based on the Hash of company_id
SELECT create_distributed_table('companies', 'company_id');
SELECT create_distributed_table('users', 'company_id');
SELECT create_distributed_table('products', 'company_id');
SELECT create_distributed_table('orders', 'company_id');

Multi-Tenant Seed Data

Now we will populate our cluster with some dummy companies along with some global order status entries, orders, and tenant users.

-- Global Order Lookups
INSERT INTO order_statuses VALUES (1, 'Pending'), (2, 'Shipped'), (3, 'Delivered');

-- Company Tenants
INSERT INTO companies VALUES (1, 'EbaMart'), (2, 'FufuHub'), (3, 'GymGear');

-- Multi-Tenant Users scattered across companies
INSERT INTO users (company_id, user_id, username) VALUES
(1, 101, 'Samuel'),   -- EbaMart Customer
(2, 201, 'Folashade'),-- FufuHub Customer
(1, 102, 'Umoh'),     -- EbaMart Customer
(3, 301, 'Oladejo'),  -- GymGear Customer
(2, 202, 'BigSam');   -- FufuHub Customer

-- Orders
INSERT INTO orders (company_id, order_id, user_id, status_id, order_amount) VALUES
(1, 5001, 101, 1, 150.00),  -- EbaMart order: Pending
(1, 5002, 102, 3, 45.00),   -- EbaMart order: Delivered
(2, 6001, 201, 2, 99.00),   -- FufuHub order: Shipped
(2, 6002, 202, 1, 250.00);  -- FufuHub order: Pending

Verify Layout and Node Placement

We can now verify the layout and node placement to see where they are added to and display it by running:

SELECT
    u.company_id,
    u.user_id,
    u.username,
    s.shardid,
    s.nodename
FROM users u
JOIN citus_shards s
  ON s.shardid = get_shard_id_for_distribution_column('users', u.company_id)
ORDER BY u.company_id;

Output:

company_id	user_id	username	shardid	nodename
1	101	Samuel	102042	db-sharding-worker-2
1	102	Umoh	102042	db-sharding-worker-2
2	201	Folashade	102065	db-sharding-worker-1
2	202	BigSam	102065	db-sharding-worker-1
3	301	Oladejo	102056	db-sharding-worker-1

As you can see, all data for company_id = 1 sits neatly on db-sharding-worker-2, while company_id = 2 sits on db-sharding-worker-1. This guarantees that queries for a single tenant will only hit one node.

We can also inspect the query plan for joining distributed tables with reference tables:

EXPLAIN (VERBOSE, COSTS OFF)
SELECT o.order_id, o.order_amount, s.status_name
FROM orders o
JOIN order_statuses s ON o.status_id = s.status_id
WHERE o.company_id = 1;

Output:

                                      QUERY PLAN
---------------------------------------------------------------------
---------------------------------------------------------------------
---------------------------------------------------------------------
------------------------
 Custom Scan (Citus Adaptive)
   Output: remote_scan.order_id, remote_scan.order_amount, remote_scan.status_name
   Task Count: 1
   Tasks Shown: All
   ->  Task
         Query: SELECT o.order_id, o.order_amount, s.status_name FROM (public.orders_102106 o JOIN public.order_statuses_102008 s ON ((o.status_id OPERATOR(pg_catalog.=) s.status_id))) WHERE (o.company_id OPERATOR(pg_catalog.=) 1)
         Node: host=db-sharding-worker-2 port=5432 dbname=postgres
         ->  Hash Join
               Output: o.order_id, o.order_amount, s.status_name
               Hash Cond: (s.status_id = o.status_id)
               ->  Seq Scan on public.order_statuses_102008 s
                     Output: s.status_id, s.status_name
               ->  Hash
                     Output: o.order_id, o.order_amount, o.status_id
                     ->  Bitmap Heap Scan on public.orders_102106 o
                           Output: o.order_id, o.order_amount, o.status_id
                           Recheck Cond: (o.company_id = 1)
                           ->  Bitmap Index Scan on orders_pkey_102106
                                 Index Cond: (o.company_id = 1)
(19 rows)

The query plan shows a Custom Scan (Citus Adaptive) delegating the query to the specific worker node holding company_id = 1.

We can check the active worker nodes with:

SELECT * FROM master_get_active_worker_nodes();

Output:

node_name	node_port
db-sharding-worker-3	5432
db-sharding-worker-1	5432
db-sharding-worker-2	5432

Rebalancing the Cluster

Anytime you dynamically scale the cluster and register a new worker node, you must run the Citus Shard Rebalancer:

SELECT citus_rebalance_start();

When new nodes are added, the coordinator does not automatically move existing data onto them. We have to explicitly balance the load. The rebalancer handles this by dynamically moving a subset of your tenants' data over to the new worker servers in the background, without taking the database offline.

Production Setup

The Docker Compose setup is perfect for local testing and development. However, running a distributed database in production across multiple physical servers or virtual machines requires a slightly different approach.

We must first change how network addressing and orchestration work. In a local setup, Docker Compose handles networking automatically because everything shares a single machine loopback. On production VMs, each container must be bound to the public or private IP address of its physical host machine, and you must enforce strict firewall security.

1. Spin up your VMs

Spin up your virtual machines on your preferred cloud provider. Assuming we provision 4 VMs, our network layout will look something like this:

• VM 1 (Coordinator Node): Private IP 10.0.0.10
• VM 2 (Worker Node 1): Private IP 10.0.0.11
• VM 3 (Worker Node 2): Private IP 10.0.0.12
• VM 4 (Worker Node 3): Private IP 10.0.0.13

I used random private IPs, but you should substitute them with your real VMs' private IPs.

2. Install Docker

Next, you will need to install Docker and Docker Compose on every VM. The installation commands will depend on your specific Linux distribution, so refer to the official Docker documentation for your OS.

You will no longer use a single master compose file for the cluster. Instead, you deploy an isolated, optimized compose layout directly on each corresponding VM instance.

3. The Coordinator Instance (VM 1)

Create a docker-compose.yml file specifically tailored to map the Postgres engine port directly out to the host VM network interface:

services:
  master:
    container_name: "citus_coordinator"
    image: "citusdata/citus:14.1.0"
    ports:
      - "5432:5432"
    restart: always
    environment:
      POSTGRES_USER: "citus_admin"
      POSTGRES_PASSWORD: "YOUR_SUPER_SECURE_PRODUCTION_PASSWORD"
      PGUSER: "citus_admin"
      PGPASSWORD: "YOUR_SUPER_SECURE_PRODUCTION_PASSWORD"
      POSTGRES_HOST_AUTH_METHOD: "scram-sha-256"
    volumes:
      - coordinator_data:/var/lib/postgresql/data

volumes:
  coordinator_data:

4. The Worker Instances (VMs 2, 3, and 4)

Every worker VM gets its own standalone compose file. Notice we expose port 5432 on the workers too, because the coordinator node needs a direct network pipe to talk to them:

services:
  worker:
    container_name: "citus_worker"
    image: "citusdata/citus:14.1.0"
    ports:
      - "5432:5432"
    restart: always
    environment:
      POSTGRES_USER: "citus_admin"
      POSTGRES_PASSWORD: "YOUR_SUPER_SECURE_PRODUCTION_PASSWORD"
      PGUSER: "citus_admin"
      PGPASSWORD: "YOUR_SUPER_SECURE_PRODUCTION_PASSWORD"
      POSTGRES_HOST_AUTH_METHOD: "scram-sha-256"
    volumes:
      - worker_data:/var/lib/postgresql/data

volumes:
  worker_data:

5. Configure Network Firewall Security

Since you are running across independent cloud instances, your cloud platform's Security Groups or Firewalls (like AWS Security Groups or ufw on Ubuntu) must be strictly configured to prevent public database scanning.

• On the Coordinator VM (VM 1): Allow inbound TCP traffic on port 5432 from Anywhere (0.0.0.0/0) only if your external application servers need public access. If your application servers are in the same VPC, restrict inbound port 5432 access exclusively to your app servers' IP subnet range.
• On the Worker VMs (VMs 2, 3, 4): Allow inbound TCP traffic on port 5432 EXCLUSIVELY from the Coordinator's Private IP (10.0.0.10). Block all other inbound network requests to the workers. They should never be exposed to the internet.

Next, log in to each VM and run the containers:

docker compose up -d

Once all containers are healthy, access the interactive postgres interface on VM 1 (the coordinator node):

docker compose exec master psql -U citus_admin

Register your external worker instances using their physical Private VPC IP addresses:

SELECT citus_add_node('10.0.0.11', 5432); -- Register Worker VM 2
SELECT citus_add_node('10.0.0.12', 5432); -- Register Worker VM 3
SELECT citus_add_node('10.0.0.13', 5432); -- Register Worker VM 4

Verify everything works:

SELECT * FROM master_get_active_worker_nodes();

6. Configure Production Password Authentication

Because we updated POSTGRES_HOST_AUTH_METHOD to scram-sha-256 for security, your instances will require passwords to authenticate cross-node.

We must add a Postgres password passfile on the coordinator container:

docker compose exec master bash -c "echo '*:*:*:citus_admin:YOUR_SUPER_SECURE_PRODUCTION_PASSWORD' >> /var/lib/postgresql/.pgpass && chmod 600 /var/lib/postgresql/.pgpass"

Surely you will know to replace YOUR_SUPER_SECURE_PRODUCTION_PASSWORD with your actual password.

Here is a detailed documentation of Citus multi-node installation and setup for reference: Citus Multi-Node Setup.

Adding Citus to an Existing Project

To add Citus to an already existing Postgres database project, the process follows these general steps:

1. Install the Extension: Install the Citus extension on your Postgres server. You will also need to add citus to the shared_preload_libraries in your postgresql.conf file and restart the database.
2. Create the Extension: Run CREATE EXTENSION citus; in your database.
3. Select a Shard Key: Identify the column that will serve as the shard key for your distributed tables, such as a company or tenant ID.
4. Convert Tables: Use create_distributed_table() to convert your existing tables into distributed tables. Use create_reference_table() for lookup tables that do not need to be sharded.
5. Migrate Data: If you already have data in the tables, it might need to be exported and loaded back in, depending on the volume of data and the specific requirements of the migration.

Conclusion

Database sharding is an incredibly powerful architectural pattern, but it comes with immense operational tradeoffs. By using tools like Citus, we can bridge the gap between standard relational SQL databases and massively distributed systems.

Whether you are starting a new multi-tenant application or scaling out an existing infrastructure, adopting a distributed mindset and orchestrating your local environments with tools like Citus and Docker Compose will save you countless hours of architectural nightmares down the road.

You can find the full Docker Compose source code for this tutorial on GitHub: umohsamuel/database-sharding.

References

• What is Database Sharding? (Google Cloud)
• What is Database Sharding? (AWS)
• Citus Docker Compose Setup

Comments

[Mads Hansen]

The single-task EXPLAIN is a good acceptance test for the distribution-key choice. I’d turn that into a schema invariant: every tenant-scoped primary key, unique constraint, and relationship should include company_id, and application queries should be tested to reject or flag plans with unexpected multi-shard task counts. Co-location only helps if tenant identity survives all the way through the key and query shape.

Also measure skew by work, not only rows. One tenant can have few rows but dominate writes, locks, or expensive queries. A production drill should include a deliberately “hot” tenant, concurrent traffic during citus_rebalance_start(), worker loss, and restore verification across coordinator plus workers. I would keep every database node on private networking and connect app servers through explicit allowlists/TLS; exposing the coordinator on 0.0.0.0/0, even conditionally, is an avoidable database boundary risk.