In Akka 24.05 we make it possible to achieve very high throughput with ordinary PostgreSQL as event storage – without causing a bottleneck in the database. What?!?! How is that possible? Akka delivers an astonishing throughput by not using the database for read requests, and automatically sharding the write requests over many databases, without any application-level concerns or specific sharding features in the databases
In today's data-driven world, where system performance and scalability are crucial for handling massive data volumes and concurrent users, the limitations of traditional database management systems are becoming increasingly evident. Once the backbone of most applications, these systems are struggling to keep up with the growing demands, especially in handling high levels of concurrent transactional operations.
Akka can handle many millions of entities, by sharding them over nodes in the Akka cluster. Each entity has an identity, and for each entity identity, there is only one instance that holds the system of record and processes the requests for that entity.
The state of the entity is kept in memory as long as it is active. This means it can serve read requests or command validation with full consistency before updating without additional reads from the database.
Many millions of entities can be sharded over many nodes in the cluster to support horizontal scalability. Akka ensures that each entity only exists as one instance in the cluster. This property of an entity has many benefits for high performance and enables a simple programming model without concurrency concerns.
There might not be room for all entities to be kept active in memory all the time and therefore entities can be passivated, for example, based on a least-recently-used strategy. When the entity is used again it recovers its state from durable storage and becomes an active entity with its system of record in memory, backed by consistent durable storage. This recovery process is also used in cases of rolling updates, rebalance, and abnormal crashes.
For durable storage, the changes of the entity are stored as events in an append-only event log. The recovery process will replay the change events to reconstruct the state. As an optimization, snapshots of the state may be stored occasionally to reduce the number of events to read and replay. In that case, the snapshot is loaded, and the events after the snapshot are replayed.
The event log is used for more than just recovering the state of the entities. From the stream of events from all entities, the application can store other representations for the purpose of queries or process the events by calling external services. The event processing can be split up over many event consumers in the cluster. This is a data partitioning approach where the events are grouped into slices based on a deterministic hash of the entity identifier that emitted the event. Each event consumer takes responsibility for a range of slices, and Akka is managing the event consumers in the cluster. The number of event consumers can be changed dynamically.
Additionally, the event log is used for communication between different services or replicas of the entities running in other cloud or edge locations.
From an operational point of view, it’s easy to use an ordinary relational database, such as PostgreSQL, for the durable storage and retrieval of events. Such regular databases also have a good price-performance ratio. However, a single database can become a bottleneck for applications that have high throughput requirements. We have seen that a single Amazon RDS PostgreSQL (8 CPUs db.r6g.2xlarge) can handle 25,000 event writes per second. What if we could use more than one database to scale for higher throughput?
We want all events for a single entity instance to be written to the same database to avoid cross-database queries when reading the events, when the entity is recovered. More difficult, and equally important, is that the reads across many entities for the event consumers should be possible with a single query from one database. Fortunately, we had precisely those properties for the slice partitioning used by projections.
Events are grouped into slices based on a deterministic hash of the persistence identifier of the entity that emitted the event. There are 1024 slices, from 0 to 1023. A projection instance consumes events from a range of slices. For example, running a projection with 4 instances the slice ranges would be 0-255, 256-511, 512-767, 768-1023. Changing to 8 projection instances means that the ranges would be 0-127, 128-255, 256-383, …, 768-895, 896-1023.
We can map such slice ranges to databases. Above diagram illustrates running a projection with 16 instances and 4 databases with certain slice ranges. Each such slice range is deterministically mapped to one database. For example entity A-3 is hashed to slice 132, which is handled by the projection instance for slice range 128-191, and that is stored in the database responsible for slice range 0-255. Each entity maps to one slice. Each of the 16 projection instances will consume events from 64 slices. Each of the 4 databases will store 256 slices.
This new feature in Akka 24.05 opens up great opportunities to use cost-efficient and simple-to-maintain event storage for massively scalable Akka applications. Early testing indicates that write throughput grows nearly linearly as additional database shards are added.
