Designing Real-Time Analytics with Database Sharding
What It Is
Database sharding is a horizontal partitioning strategy that distributes data across multiple independent database instances, called shards. Each shard holds a subset of the data and operates as a standalone database. When combined with real-time analytics, sharding enables high-throughput ingestion, low-latency queries, and linear scalability for time-sensitive analytical workloads such as dashboards, monitoring systems, fraud detection, and IoT telemetry.
In a real-time analytics context, sharding allows you to parallelize both writes and reads across many nodes. Instead of a single database struggling to handle millions of events per second, each shard handles a fraction of the load, making it possible to query recent data with sub-second response times even as data volumes grow into the petabytes.
-- Conceptual shard layout for real-time events
-- Shard 1: events with user_id hash % 4 == 0
-- Shard 2: events with user_id hash % 4 == 1
-- Shard 3: events with user_id hash % 4 == 2
-- Shard 4: events with user_id hash % 4 == 3
-- Each shard contains identical table schemas
CREATE TABLE events (
event_id UUID PRIMARY KEY,
user_id INT NOT NULL,
event_type VARCHAR(50) NOT NULL,
event_data JSONB,
occurred_at TIMESTAMPTZ NOT NULL,
ingested_at TIMESTAMPTZ DEFAULT NOW()
);
-- Index for real-time range queries
CREATE INDEX idx_events_occurred_at ON events (occurred_at DESC);
CREATE INDEX idx_events_user_id ON events (user_id);Why It Matters
Real-time analytics systems face three fundamental challenges that sharding directly addresses:
- Write throughput bottleneck: A single database node has finite I/O and CPU capacity. At millions of writes per second, it becomes saturated. Sharding distributes writes across many nodes, allowing aggregate throughput to scale linearly with the number of shards.
- Query latency under large datasets: Scanning billions of rows on a single node is slow even with indexes. Sharding reduces the data volume per node, enabling faster scans and aggregations for time-windowed queries.
- Hotspot mitigation: Without sharding, a sudden spike in activity for a popular user or device can overwhelm a single node. Sharding spreads activity across nodes, preventing any single shard from becoming a bottleneck.
Real-world impact: Companies like Uber, Discord, and Slack rely on sharded databases to power their real-time analytics pipelines. For example, Uber's Apache Cassandra-based analytics platform ingests billions of events per day and serves dashboard queries with p99 latency under 100 milliseconds — a feat impossible without sharding.
How to Use It: A Step-by-Step Guide
1. Choose a Sharding Key
The sharding key determines how data is distributed across shards. For real-time analytics, the key must support both balanced distribution and efficient query routing. Common choices include:
- User ID or Tenant ID: Good for multi-tenant analytics where queries are scoped to a single tenant.
- Hash of a natural key: Ensures even distribution regardless of key distribution.
- Time-based + hash combination: Useful when you need time-range queries across all shards.
// Sharding key selection in a Node.js proxy layer
function getShardId(userId, numShards) {
// Consistent hashing using a fast hash function
const hash = require('crypto')
.createHash('md5')
.update(String(userId))
.digest('hex');
// Take first 4 bytes, convert to integer, mod by shard count
const intHash = parseInt(hash.substring(0, 8), 16);
return intHash % numShards;
}
// Usage during write
const shardId = getShardId(event.userId, 8);
const shardConnection = shardPool[shardId];
await shardConnection.query(
'INSERT INTO events (event_id, user_id, event_type, event_data, occurred_at) VALUES ($1, $2, $3, $4, $5)',
[event.eventId, event.userId, event.eventType, event.eventData, event.occurredAt]
);2. Design the Shard Topology
Decide on the number of shards and their physical deployment. For real-time workloads, start with more shards than you think you need to accommodate future growth without resharding.
// Shard topology configuration (YAML-based config)
shards:
total: 16
replication_factor: 3
nodes:
- shard_id: 0
host: shard0-db1.example.com
port: 5432
replicas:
- shard0-db2.example.com
- shard0-db3.example.com
- shard_id: 1
host: shard1-db1.example.com
port: 5432
replicas:
- shard1-db2.example.com
- shard1-db3.example.com
# ... continues for all 16 shards
# Connection pool configuration per shard
connection_pool:
min_size: 10
max_size: 50
idle_timeout_ms: 30000
connection_timeout_ms: 50003. Implement the Query Router
The query router is the brain of the sharded system. It parses incoming queries, determines which shards to target, sends the queries in parallel, and merges the results.
// Query router for real-time analytics
class ShardedQueryRouter {
constructor(shardPool) {
this.shardPool = shardPool;
}
// Route a time-range aggregation query across all shards
async queryTimeSeries(eventType, startTime, endTime, granularity) {
const numShards = this.shardPool.length;
const promises = [];
for (let shardId = 0; shardId < numShards; shardId++) {
const query = `
SELECT
date_trunc('${granularity}', occurred_at) AS bucket,
COUNT(*) AS event_count,
AVG(EXTRACT(EPOCH FROM (ingested_at - occurred_at))) AS avg_latency_seconds
FROM events
WHERE event_type = $1
AND occurred_at BETWEEN $2 AND $3
GROUP BY bucket
ORDER BY bucket
`;
promises.push(
this.shardPool[shardId].query(query, [eventType, startTime, endTime])
);
}
// Wait for all shards and merge results
const shardResults = await Promise.all(promises);
return this.mergeTimeSeries(shardResults.map(r => r.rows));
}
// Merge sorted time buckets from multiple shards
mergeTimeSeries(shardRows) {
const merged = new Map();
for (const rows of shardRows) {
for (const row of rows) {
const key = row.bucket.toISOString();
if (!merged.has(key)) {
merged.set(key, { bucket: row.bucket, count: 0, totalLatency: 0, countLatency: 0 });
}
const entry = merged.get(key);
entry.count += parseInt(row.event_count, 10);
entry.totalLatency += parseFloat(row.avg_latency_seconds) * parseInt(row.event_count, 10);
entry.countLatency += parseInt(row.event_count, 10);
}
}
// Convert to sorted array and compute final averages
return Array.from(merged.values())
.sort((a, b) => a.bucket - b.bucket)
.map(entry => ({
bucket: entry.bucket,
event_count: entry.count,
avg_latency_seconds: entry.countLatency > 0
? entry.totalLatency / entry.countLatency
: 0
}));
}
// Single-shard query for user-scoped analytics
async queryUserEvents(userId, limit = 100) {
const shardId = getShardId(userId, this.shardPool.length);
const query = `
SELECT event_id, event_type, event_data, occurred_at
FROM events
WHERE user_id = $1
ORDER BY occurred_at DESC
LIMIT $2
`;
const result = await this.shardPool[shardId].query(query, [userId, limit]);
return result.rows;
}
}4. Build the Ingestion Pipeline
Real-time analytics requires high-throughput ingestion. Use a streaming buffer (Apache Kafka, Redpanda, or NATS) to decouple producers from the database, then batch-write to each shard.
// Kafka consumer that batches events and writes to shards
const { Kafka } = require('kafkajs');
class ShardedIngestionWorker {
constructor(kafkaConfig, shardPool, batchSize = 1000, flushIntervalMs = 100) {
this.shardPool = shardPool;
this.batchSize = batchSize;
this.flushIntervalMs = flushIntervalMs;
this.buffers = new Map(); // shardId -> array of events
this.consumer = new Kafka(kafkaConfig).consumer({ groupId: 'analytics-ingester' });
}
async start() {
await this.consumer.connect();
await this.consumer.subscribe({ topic: 'raw-events', fromBeginning: false });
// Flush buffers periodically
setInterval(() => this.flushAll(), this.flushIntervalMs);
await this.consumer.run({
eachMessage: async ({ message }) => {
const event = JSON.parse(message.value.toString());
const shardId = getShardId(event.userId, this.shardPool.length);
if (!this.buffers.has(shardId)) {
this.buffers.set(shardId, []);
}
this.buffers.get(shardId).push(event);
if (this.buffers.get(shardId).length >= this.batchSize) {
await this.flushShard(shardId);
}
}
});
}
async flushShard(shardId) {
const events = this.buffers.get(shardId);
if (!events || events.length === 0) return;
this.buffers.set(shardId, []); // Clear before write to avoid re-entry
const client = this.shardPool[shardId];
// Use a single multi-row INSERT for efficiency
const placeholders = events.map((_, i) => {
const base = i * 5;
return `($${base + 1}, $${base + 2}, $${base + 3}, $${base + 4}, $${base + 5})`;
}).join(', ');
const values = events.flatMap(e => [
e.eventId, e.userId, e.eventType,
JSON.stringify(e.eventData), e.occurredAt
]);
const query = `INSERT INTO events (event_id, user_id, event_type, event_data, occurred_at) VALUES ${placeholders}`;
await client.query(query, values);
}
async flushAll() {
const shardIds = Array.from(this.buffers.keys());
await Promise.all(shardIds.map(id => this.flushShard(id)));
}
}5. Implement Aggregation and Materialized Views
For real-time dashboards, pre-aggregate data at write time to avoid expensive scans at query time. Use per-shard materialized views or a dedicated aggregation table.
-- Per-shard materialized view for minute-level aggregations
CREATE MATERIALIZED VIEW events_minute_agg
AS
SELECT
date_trunc('minute', occurred_at) AS bucket,
event_type,
COUNT(*) AS event_count,
COUNT(DISTINCT user_id) AS unique_users,
AVG(EXTRACT(EPOCH FROM (ingested_at - occurred_at))) AS avg_ingestion_latency
FROM events
GROUP BY bucket, event_type
WITH DATA;
-- Refresh it periodically or on demand
CREATE OR REPLACE FUNCTION refresh_minute_agg()
RETURNS void AS $$
BEGIN
REFRESH MATERIALIZED VIEW CONCURRENTLY events_minute_agg;
END;
$$ LANGUAGE plpgsql;
-- For real-time needs, also maintain a "hot" table for the last 5 minutes
CREATE TABLE events_hot_minute (
bucket TIMESTAMPTZ NOT NULL,
event_type VARCHAR(50) NOT NULL,
event_count INT DEFAULT 0,
unique_users INT DEFAULT 0,
avg_ingestion_latency DOUBLE PRECISION DEFAULT 0,
PRIMARY KEY (bucket, event_type)
);
-- Upsert into hot table on each batch write
-- (called from the ingestion worker after flush)
async function updateHotAggregates(shardClient, events) {
for (const event of events) {
const bucket = new Date(event.occurredAt);
bucket.setSeconds(0, 0); // truncate to minute
const query = `
INSERT INTO events_hot_minute (bucket, event_type, event_count, unique_users, avg_ingestion_latency)
VALUES ($1, $2, 1, 1, EXTRACT(EPOCH FROM (NOW() - $3)))
ON CONFLICT (bucket, event_type)
DO UPDATE SET
event_count = events_hot_minute.event_count + 1,
unique_users = events_hot_minute.unique_users + 1,
avg_ingestion_latency = (events_hot_minute.avg_ingestion_latency * events_hot_minute.event_count + EXTRACT(EPOCH FROM (NOW() - $3))) / (events_hot_minute.event_count + 1)
`;
await shardClient.query(query, [bucket, event.eventType, event.occurredAt]);
}
}6. Build the Real-Time Dashboard API
The API layer queries all shards in parallel and combines results. For time-critical dashboards, query the hot aggregation tables and fall back to the materialized views for historical data.
// Real-time dashboard API endpoint
class AnalyticsAPI {
constructor(router) {
this.router = router;
}
async getRealtimeMetrics(req, res) {
const { eventType, windowMinutes = 5 } = req.query;
const endTime = new Date();
const startTime = new Date(endTime.getTime() - windowMinutes * 60 * 1000);
try {
// Query hot aggregates from all shards
const numShards = this.router.shardPool.length;
const promises = [];
for (let shardId = 0; shardId < numShards; shardId++) {
const query = `
SELECT bucket, event_type, event_count, unique_users, avg_ingestion_latency
FROM events_hot_minute
WHERE event_type = $1
AND bucket BETWEEN $2 AND $3
ORDER BY bucket
`;
promises.push(
this.router.shardPool[shardId].query(query, [eventType, startTime, endTime])
);
}
const results = await Promise.all(promises);
const merged = this.mergeHotMetrics(results.map(r => r.rows));
res.json({
status: 'ok',
window_minutes: parseInt(windowMinutes, 10),
metrics: merged
});
} catch (err) {
res.status(500).json({ status: 'error', message: err.message });
}
}
mergeHotMetrics(shardRows) {
const map = new Map();
for (const rows of shardRows) {
for (const row of rows) {
const key = `${row.bucket.toISOString()}|${row.event_type}`;
if (!map.has(key)) {
map.set(key, { ...row });
} else {
const existing = map.get(key);
existing.event_count += row.event_count;
existing.unique_users += row.unique_users;
existing.avg_ingestion_latency =
(existing.avg_ingestion_latency + row.avg_ingestion_latency) / 2;
}
}
}
return Array.from(map.values()).sort((a, b) => a.bucket - b.bucket);
}
async getUserTimeline(req, res) {
const { userId, limit = 50 } = req.params;
try {
const events = await this.router.queryUserEvents(parseInt(userId, 10), parseInt(limit, 10));
res.json({ status: 'ok', userId, events });
} catch (err) {
res.status(500).json({ status: 'error', message: err.message });
}
}
}Best Practices
1. Choose the Right Sharding Strategy
- Consistent Hashing: Best for dynamic shard counts. Minimizes data movement when adding or removing shards.
- Range-Based Sharding: Simple but prone to hotspots if data distribution is uneven. Avoid for real-time workloads unless you have natural range boundaries like geographic regions.
- Directory-Based Sharding: Uses a lookup table to map keys to shards. Flexible but introduces a single point of failure and latency overhead.
// Consistent hashing with virtual nodes for even distribution
class ConsistentHashRing {
constructor(numVirtualNodes = 150) {
this.numVirtualNodes = numVirtualNodes;
this.ring = new Map(); // position -> shardId
this.sortedPositions = [];
}
addShard(shardId) {
for (let i = 0; i < this.numVirtualNodes; i++) {
const hash = this.hash(`${shardId}:${i}`);
this.ring.set(hash, shardId);
this.sortedPositions.push(hash);
}
this.sortedPositions.sort((a, b) => a - b);
}
removeShard(shardId) {
for (let i = 0; i < this.numVirtualNodes; i++) {
const hash = this.hash(`${shardId}:${i}`);
this.ring.delete(hash);
}
this.sortedPositions = Array.from(this.ring.keys()).sort((a, b) => a - b);
}
getShard(key) {
if (this.sortedPositions.length === 0) return null;
const hash = this.hash(String(key));
// Binary search for the first position >= hash
let lo = 0, hi = this.sortedPositions.length - 1;
while (lo < hi) {
const mid = Math.floor((lo + hi) / 2);
if (this.sortedPositions[mid] >= hash) {
hi = mid;
} else {
lo = mid + 1;
}
}
const position = this.sortedPositions[lo];
return this.ring.get(position) || this.ring.get(this.sortedPositions[0]);
}
hash(str) {
let h = 0;
for (let i = 0; i < str.length; i++) {
h = ((h << 5) - h) + str.charCodeAt(i);
h |= 0; // convert to 32-bit integer
}
return h >>> 0; // ensure positive
}
}2. Design for Cross-Shard Queries
Not all queries can be routed to a single shard. For real-time analytics that span all users or all event types, you must query all shards and merge results. Mitigate the overhead by:
- Pre-aggregating data per shard so that cross-shard queries only need to fetch small aggregation results, not raw rows.
- Using scatter-gather patterns with parallel async queries and a timeout to prevent slow shards from blocking the entire query.
- Caching merged results at the API layer for frequently accessed time windows (e.g., last 5 minutes).
// Scatter-gather with timeout and partial results
async function scatterGather(shardPool, query, params, timeoutMs = 5000) {
const numShards = shardPool.length;
const results = new Array(numShards).fill(null);
const promises = shardPool.map((shard, index) =>
Promise.race([
shard.query(query, params).then(result => { results[index] = result.rows; }),
new Promise((_, reject) =>
setTimeout(() => reject(new Error(`Shard ${index} timed out`)), timeoutMs)
)
]).catch(err => {
console.warn(`Shard ${index} failed: ${err.message}`);
results[index] = []; // Return empty for failed shard
})
);
await Promise.allSettled(promises);
return results.flat(); // Return partial results even if some shards failed
}3. Monitor Shard Health and Balance
Uneven data distribution or query load can create hotspots. Monitor these metrics per shard:
- Row count and data size — detect skew in data distribution.
- Query latency (p50, p95, p99) — identify slow shards.
- Write throughput (events/sec) — ensure even ingestion load.
- Connection pool utilization — prevent connection starvation.
-- Monitoring query for shard balance
SELECT
shard_id,
COUNT(*) AS row_count,
pg_size_pretty(SUM(pg_total_relation_size('events'))) AS total_size,
MIN(occurred_at) AS oldest_event,
MAX(occurred_at) AS newest_event,
COUNT(*) FILTER (WHERE occurred_at > NOW() - INTERVAL '5 minutes') AS events_last_5min
FROM events
GROUP BY shard_id
ORDER BY shard_id;
-- Expected output for a well-balanced 4-shard system:
-- shard_id | row_count | total_size | oldest_event | newest_event | events_last_5min
-- ---------+-----------+------------+----------------------+----------------------+-----------------
-- 0 | 2500123 | 1.2 GB | 2024-01-01 00:00:00 | 2024-09-15 12:34:56 | 45231
-- 1 | 2498876 | 1.2 GB | 2024-01-01 00:00:01 | 2024-09-15 12:34:56 | 45198
-- 2 | 2501345 | 1.2 GB | 2024-01-01 00:00:00 | 2024-09-15 12:34:55 | 45302
-- 3 | 2499656 | 1.2 GB | 2024-01-01 00:00:02 | 2024-09-15 12:34:56 | 451674. Plan for Resharding
As data grows, you will eventually need to add or remove shards. Design for minimal downtime:
- Use consistent hashing to minimize data movement when adding shards (only a fraction of data moves).
- Implement dual-write during resharding: write to both old and new shards, then switch reads after verification.
- Use a proxy layer that abstracts shard topology from application code, allowing topology changes without code changes.
// Proxy layer that supports dynamic shard topology
class ShardProxy {
constructor(initialTopology) {
this.topology = initialTopology;
this.ring = new ConsistentHashRing();
for (const shard of this.topology) {
this.ring.addShard(shard.id);
}
}
async write(event) {
const shardId = this.ring.getShard(event.userId);
const shard = this.topology.find(s => s.id === shardId);
// Write to primary shard
await shard.client.query('INSERT INTO events ...', [event]);
// If dual-write mode is active, also write to new shard
if (this.dualWriteTarget) {
await this.dualWriteTarget.client.query('INSERT INTO events ...', [event]);
}
}
async read(userId) {
const shardId = this.ring.getShard(userId);
const shard = this.topology.find(s => s.id === shardId);
return shard.client.query('SELECT * FROM events WHERE user_id = $1', [userId]);
}
beginResharding(newShards) {
this.dualWriteTarget = newShards[0]; // simplified for illustration
// After dual-write is stable, rebuild the ring and switch reads
setTimeout(() => {
this.ring = new ConsistentHashRing();
for (const shard of newShards) {
this.ring.addShard(shard.id);
}
this.topology = newShards;
this.dualWriteTarget = null;
}, 60000); // 1 minute dual-write window
}
}5. Optimize for Real-Time Query Patterns
Real-time analytics queries typically focus on recent data (last minutes to hours). Optimize for this access pattern:
- Use time-based partitioning within each shard (e.g., daily or hourly partitions) so that queries scanning recent data only touch a small subset of files.
- Keep hot data in memory by configuring sufficient shared buffers and using pg_prewarm or similar tools.
- Use covering indexes for common aggregation queries to avoid heap lookups.
-- Time-based partitioning within each shard (PostgreSQL declarative partitioning)
CREATE TABLE events (
event_id UUID NOT NULL,
user_id INT NOT NULL,
event_type VARCHAR(50) NOT NULL,
event_data JSONB,
occurred_at TIMESTAMPTZ NOT NULL,
ingested_at TIMESTAMPTZ DEFAULT NOW()
) PARTITION BY RANGE (occurred_at);
-- Create daily partitions for recent data
CREATE TABLE events_2024_09_15 PARTITION OF events
FOR VALUES FROM ('2024-09-15') TO ('2024-09-16');
CREATE TABLE events_2024_09_16 PARTITION OF events
FOR VALUES FROM ('2024-09-16') TO ('2024-09-17');
-- Create a default partition for future dates
CREATE TABLE events_future PARTITION OF events DEFAULT;
-- Covering index for the most common real-time query
CREATE INDEX idx_events_realtime ON events (event_type, occurred_at DESC)
INCLUDE (user_id, event_data);6. Handle Failures Gracefully
In a sharded system, individual shard failures should not bring down the entire analytics platform. Implement these patterns:
- Circuit breakers: If a shard is slow or failing, stop sending requests to it and return cached or degraded results.
- Replica reads: For read-heavy workloads, route read queries to replica shards while writes go to the primary.
- Dead shard detection: Proactively detect and remove unresponsive shards from the routing pool.
// Circuit breaker for shard connections
class CircuitBreaker {
constructor(failureThreshold = 5, recoveryTimeoutMs = 30000) {
this.failureThreshold = failureThreshold;
this.recoveryTimeoutMs = recoveryTimeoutMs;
this.failureCount = 0;
this.state = 'CLOSED'; // CLOSED, OPEN, HALF_OPEN
this.lastFailureTime = null;
}
async call(fn) {
if (this.state === 'OPEN') {
if (Date.now() - this.lastFailureTime > this.recoveryTimeoutMs) {
this.state = 'HALF_OPEN';
} else {
throw new Error('Circuit breaker is OPEN');
}
}
try {
const result = await fn();
if (this.state === 'HALF_OPEN') {
this.state = 'CLOSED';
this.failureCount = 0;
}
return result;
} catch (err) {
this.failureCount++;
this.lastFailureTime = Date.now();
if (this.failureCount >= this.failureThreshold) {
this.state = 'OPEN';
}
throw err;
}
}
}
// Usage in the query router
class ResilientShardPool {
constructor(shardConnections) {
this.shards = shardConnections.map((conn, i) => ({
id: i,
conn,
breaker: new CircuitBreaker()
}));
}
async query(shardId, query, params) {
const shard = this.shards[shardId];
return shard.breaker.call(() => shard.conn.query(query, params));
}
}Putting It All Together: End-to-End Architecture
A complete real-time analytics system with database sharding combines all the components discussed above. Here is a reference architecture:
// End-to-end system composition
class RealtimeAnalyticsSystem {
constructor(config) {
// 1. Shard topology
this.shardPool = this.initializeShardPool(config.shards);
// 2. Consistent hashing ring
this.hashRing = new ConsistentHashRing();
for (let i = 0; i < this.shardPool.length; i++) {
this.hashRing.addShard(i);
}
// 3. Query router with circuit breakers
this.router = new ShardedQueryRouter(
new ResilientShardPool(this.shardPool)
);
// 4. Ingestion pipeline
this.ingestionWorker = new ShardedIngestionWorker(
config.kafka,
this.shardPool,
config.batchSize,
config.flushIntervalMs
);
// 5. API layer
this.api = new AnalyticsAPI(this.router);
// 6. Health monitor
this.monitor = new ShardMonitor(this.shardPool);
}
async start() {
await this.ingestionWorker.start();
this.monitor.start(60000); // check every 60 seconds
// Start HTTP server for dashboard API
const express = require('express');
const app = express();
app.get('/api/metrics', (req, res) => this.api.getRealtimeMetrics(req, res));
app.get('/api/user/:userId/events', (req, res) => this.api.getUserTimeline(req, res));
app.get('/api/health', (req, res) => this.monitor.getHealth(req, res));
app.listen(8080, () => console.log('Analytics API listening on port 8080'));
}
}
// Boot the system
const config = {
shards: [
{ host: 'shard0.example.com', port: 5432, database: 'analytics' },
{ host: 'shard1.example.com', port: 5432, database: 'analytics' },
{ host: 'shard2.example.com', port: 5432, database: 'analytics' },
{ host: 'shard3.example.com', port: 5432, database: 'analytics' },
],
kafka: {
brokers: ['kafka1.example.com:9092', 'kafka2.example.com:9092'],
clientId: 'analytics-ingester'
},
batchSize: 1000,
flushIntervalMs: 100
};
const system = new RealtimeAnalyticsSystem(config);
system.start().catch(console.error);Conclusion
Designing a real-time analytics system with database sharding is a powerful approach to achieving horizontal scalability, low-latency queries, and high ingestion throughput at any data scale. By carefully selecting a sharding key that aligns with your query patterns, implementing a robust query router with scatter-gather capabilities, pre-aggregating data at write time, and building resilience through circuit breakers and health monitoring, you can create an analytics platform that delivers sub-second insights even as data volumes grow into the billions of events per day. The patterns and code examples provided in this tutorial give you a production-ready foundation that you can adapt to your specific use case, whether you are building a real-time dashboard, a fraud detection system, or an IoT telemetry pipeline. Remember that sharding introduces operational complexity, so invest in monitoring, automation, and gradual rollout strategies to ensure your system remains reliable and maintainable as it scales.