Transcription
Notion is one of the fastest-growing softwares ever created. But how can you store billions of database entries fast?
If you're new here, my name is Lewis. I'm on a mission to inspire developers and tech enthusiasts. Let's go. So how does Notion work under the hood? Well, every time you open up a new Notion page, each block is rendered in a separate database row. So, for example, a checkbox will have an ID that's unique to every block ever, the type of block that it is, the child blocks that it is referenced by ID, as well as the parent block that is referenced by an ID. And fun fact, actually, if you go into Notion and use the arrow keys, it will do it by blocks. So any decent-sized page can have like 200 to 400 blocks.
But like, just think about this for one second. There were 1,000 users in 2020. Well, four years later, that number is now 100,000,000 users. All creating tens of thousands of blogs each. And just remember, Notion is online only, so if they broke down, everybody loses access to their notes.
In 2021, Notion was using a single Postgres database to handle all of this. So once they hit about 20,000,000,000 blocks, things started to slow down and deteriorate. They needed effects really fast. And after a huge consideration, they decided to do something called sharding. A common way to buff up your database is to continuously buff up the host machine that is on, having a larger and larger machine to deal with more and more traffic. But instead, with sharding, you would take that database and split it amongst much, much smaller machines. This will automatically make the traffic smaller per machine and also reduce heavy queries. But one downside to this is that you have to write your own application code to determine where users A's data is being held.
So the engineers at Notion looked at their block model and decided to shard everything related to this block table, like the workspace ID, discussions, comments, etc. They then decided to divide them up based on the workspace ID, which every block can only be a child of one at a time. So after crunching all of the numbers, this is what they came up with. They will turn their one giant database into 32 separate database instances, all hosting 15 shards within them, meaning 480 logical shards.
But how do they do all of this? Well, they used a combination of things, but one being the double write method which I actually talked about in another video about how Uber handles trillions of transactions. The double write method writes the same data to the new database as well as the old, and since Notion had a hard time keeping up, they created an audit log that a catch-up worker would go through periodically and add in later. Then, to fill in all of the old data, they used a machine with 96 CPUs that took three days to backfill the entire Notion database. After verification, Notion went down for five minutes to switch over systems. And ta-da, that was very fast.
But one issue came up that was becoming very obvious. Postgres was being used everywhere for everything. The same infrastructure was being used for both online users and offline abilities like machine learning and analytics. This is where a data lake comes into play here. They extract the data from the Postgres shards, load the raw data from the shards into the Snowflake data lake using Fivetran, and then transform the data into a single table where analytics can be crunched. Extract, load, and transform.
But what even is a data lake? A data lake is a centralized repository for a huge amount of data. This is often used for developers when they want to dump things in places and use at a later time. It's kind of like polluting a lake to only clean it later. You know what? I don't really think that's a great analogy. For analytics, you can throw in raw data like text files, logs, and process it into useful insights later. And the best part is that it's much cheaper than traditional storage methods.
But remember, Notion was doubling the amount of blocks being created every six to twelve months. So this system that they had already was becoming an issue rapidly. Think about how we use note applications or Word documents. It's often an iterative process, making mistakes, thinking of a better line later, changing the subject matter, and much more. For a lot of blog-like infrastructures, you would read a single database row, change the text, and hit update, meaning one update is happening throughout the entire transaction. But with Notion, I might be doing 200 updates across many database rows all in real time. But Snowflake was best at handling brand new inserts rather than updates. It also just became super expensive to do certain computations with their existing data models. So they said screw it and decided to build their own data lake designed specifically for Notion. But they had to act fast because users were continuing to double. The goal was to create a system that could store both raw and processed data, work with Notion's update-heavy block data extremely fast, and then use modern capabilities like AI, search, and more that require unstructured data.
And this is the system that they finalized on. They created their central repository with Amazon S3 since they were already using AWS for the rest of their infrastructure. That way they could use other services like Elasticsearch, vector databases, and a key-value store. To handle the transform aspects of the ELT process, they decided to use Apache Spark. Spark processes large datasets across many clusters of computers, all in memory. Spark also has a lot of ways to compute complex queries that aren't SQL based. Between Postgres and Apache Spark, we have Apache Kafka, which sends large amounts of data consistently. But before it goes to Kafka, a changed data capture is implemented to publish the incrementally changed Postgres data to Kafka. Then, to put everything together, they implemented Apache Hoodie before being put into S3, which simplifies the process of building and managing data pipelines on top of data lakes, kind of like what they're doing right now.
If you noticed, a huge part of this stack that they're working with is open-source software that you, yourself, the viewer, can go download and build free of cost. This saved Notion millions and millions of dollars and allowed them to keep up with the six to twelve-month doubling rate.
All of this was great until two years later. Shards were hitting 90% utilization rates. Things were unstable, and pgbouncer was hitting connection limits. Pgbouncer sits in between the server and the database, and it pools connections so that it can use the database connections more efficiently. Doing one-to-one direct connections would cause a huge queue to pile up causing downtime. The engineers realized that they already did the right thing two years ago. They just had to redo it. They will take 32 database machines and triple it to 96 machines. This means that the 15 shards per machine would go down to five shards per machine. To sync their old system with this new system, they used the built-in Postgres logical replication to copy over the old data and the new changes to a new database. Postgres is just really the go, let's be real here. They then built tools that would organize the incoming data into groups and make sure they were being pushed to the right databases. Then, pgbouncer would distribute traffic amongst the new 64 shards and the old 32 shards. This solved one issue, but there was another major issue that would have happened. Pgbouncer was already hitting the connection limits as is, and now they want to 3x the output? Don't think so. So the solution was sharding again. They would put their pgbouncer into four groups that would each manage 24 databases. This would then stop the connection limits from getting hit and isolate issues easier.
So everything was all set and ready to go, but it was time to do the transition. First, they tested the new system with dark reads, which would grab the same request from both databases and see if they matched. Everything looked great, but now it's time to deploy. They went one database at a time doing these steps: one, stop accepting connections; two, verify that the new database hasn't lost any data; three, update pgbouncer to point to the new databases; and four, resume traffic to the new shards. One by one, each database transitioned and went live without any downtime for users.
For solo developers, a lot of us look at the softwares that we use daily and think, jeez, I could probably build that much faster than their entire team, which, you know, probably is true, but building at scale is always hard and needs a team. I have listed the engineers from the blog articles I have grabbed all of these from down here. Thank you so much to the Notion team for sharing all of their insights on how they did it. So if you want more detailed information and the graphs that they used, make sure you click the links below. Seriously, it's some of the best things that you can read as an engineer. Let me know what company you'd like me to cover next. Peace out, Coders.