Transcript
Rajasekhar Ummadisetty: I'm Raj Ummadisetty, and joining me today is Ken Kurzweil. We are both software engineers on the Data Platform Org at Netflix. Within data platform, I am from the online datastores and Ken is from the data movement. Today, we are excited to share a story of massive architectural pivot that transformed how we move data at Netflix. Specifically, we'll walk you through our CloudStream effort to efficiently move terabytes of data from offline data systems to online serving systems. By offline data systems, we refer to our data warehouses, where data is not accessed by the devices such as televisions during the runtime. In contrast, the online data systems sit in the critical path and are serving data with high speed and random access. Achieving this transformation required us to fundamentally rethink our architecture and core principles. As a result, we were able to achieve or reduce data deployment time by 90%, and cut cost by 70%.
Outline
We'll start by discussing the core philosophy that drove every technical and strategic decision you are about to hear, confidence is currency. We'll show you how this principle anchored in safety, observability, and validation guided our actions at every step. We'll then introduce the critical architectural foundation that enabled this work, our abstractions layer, focusing on the key-value abstraction and how it was established using the principles of safety, observability, and validation. This sets the stage for the big problem, moving massive datasets from the offline data systems to online data systems. We'll then walk you through the architectural vision and the innovative solution we implemented for our first subset of the use cases, the immutable datasets. We'll then cover the complex architectural transformation that required us to transform our key-value abstraction from a stateless system into a stateful one, which helped in delivering our massive gains. We'll generalize this entire process into a repeatable capture, conversion, deployment framework. Finally, give you a peek at our future, and wrap with our key takeaways.
Confidence is Currency
Confidence is currency. It's the foundation of trust, decision-making, and progress in any organization. Individual confidence alone is not enough. What truly matters is earning the confidence of your stakeholders. Stakeholders become more confident when you demonstrate in our architecture, safety, your system can protect itself and its users. For example, by throttling the operations or handling errors gracefully. Observability, the system provides the ability to monitor, measure, and understand its internal state and progress. Validation, the system can clearly prove that it has achieved its intended results. By focusing on these principles, you build the confidence of your stakeholders that need to support your methods and goals. At Netflix, applications access database through a layer of indirection. This is a crucial and intentional architectural choice that we call as abstractions. For example, of our entire Cassandra fleet at Netflix, only 16% of our fleet is directly accessed by the applications today.
This number continues to decrease. The vast majority of our fleet is accessed through our abstractions. We handle around 70 million QPS across the entire fleet of abstractions. The largest abstraction handles around 8 million QPS. An abstraction can be backed by multiple Cassandra clusters, and the busiest Cassandra cluster handles around 1.8 million reads per second and 4.9 million writes per second.
Abstractions
Now let's walk through some of the challenges our application teams faced outside of their core business problems that we thoughtfully solved using abstractions. With direct access, each application team has to understand the specific database technology and the constant changes happening to those technologies. With abstraction, they get a stable and uniform interface. Application teams now can entirely focus on solving their core business problems and not have to worry about choosing the right database. With direct access, each application team for their access patterns has to understand how to provision for scale, when to add caching, and also be experts at cluster management. The abstraction layer allows us to optimize for access patterns and manage the cluster centrally. If we notice a use case to be a read heavy, we can provision a dedicated caching cluster to optimize the resource consumption. Each application team, also without abstractions, had to solve how to integrate with the internal systems, such as custom AuthN and AuthZ mechanisms for the database that they are integrating with.
With abstraction, data platform can solve this centrally and give it for all the applications, or make it available for all the applications. Each database also has certain limitations. For instance, if you take Cassandra, if you send large data blobs, the cluster can become unhappy. With abstractions, we can solve these limitations efficiently. For instance, in case of these large data blobs, we efficiently solve this using a mechanism called chunking. With chunking, you break these large objects into smaller objects and distribute them across the cluster. At Netflix, we have multiple abstractions to address the needs of our application teams. For applications, they want to store the key-value data, we have a key-value abstraction. If they want to store documents, we have entity. If the data is temporal in nature, we have time series. If they want to store property graphs, we have GraphKV, and many more. I've included a link to our blog post here that goes in detail on the foundation architecture that we build that enables us to build all of these abstractions on top of it.
We'll next focus on one particular abstraction, KV, and the story behind its adoption. The key-value abstraction operates as a distributed HashMap as a service. It's a two-level map, the outer level map of type string, which we call it as ID, maps to a sorted map of key-value pairs, which we call as items. The high-level architecture consists of a stateless server that connects to the control plane, fetches the namespace information, initializes the connection to the databases. The application then can use our high-level client, which offers features like SLO-based hedging, near caching, and also abstracts away low-level protocol details, such as the chunking that I just described. Or the applications can also use the low-level client directly, send a request to the service, the service transforms this request into a database query, execute it and get the response back to the application. I've included a link to the blog post here on the key-value abstraction, which goes in detail on the architecture and the feature it provides.
We did not always have abstractions. Application teams used to integrate with databases directly, but this architectural freedom collided with the reality that change is the only constant. The critical moment came when Cassandra announced that starting version 3.0, they would be deprecating the Thrift protocol and only supporting the SQL protocol. This meant all the application teams that were using or connecting to the Cassandra cluster directly using Thrift would have to do a migration. This was a massive company-wide migration. As data platform, we realized that this is not the only time a core technology would change. We would want to protect our application teams from future such disruptive migrations. We were able to convince majority of our application teams to do one last migration, update their application logic from the Thrift protocol to using the key-value interface, because for this and future such migrations, we would handle it for them.
The API migration was the first step. We focused on our three core tenets that I just described, safety, observability, and validation. They received immediate feedback that the system worked. The application teams were able to validate both the functional and non-functional requirements right away, as the transition to the KV did not require an initial data migration. This allowed them to develop, experiment, and test with their actual data, all of it without any risk. They received immediate feedback, what worked and what did not. This was without any changes. Even when committing and deploying to production, they had assurance that they could easily roll back if any issues happened. This was a true two-way door process, allowing easy rollback if any issues happened, and observed safety and observability. They exactly knew where they were in the process, and validation. The application teams could directly validate using their own data.
This process completed the API migration. The application teams were now connecting to the database using our abstraction layer, key-value abstraction. Next step was to migrate the clusters. We began by doing dual writes. The primary cluster, the old cluster continues to acknowledge the writes. We shadowed the writes to the secondary. In this mode, we got crucial signals such as the scaling needs, potential bugs, potential regressions, allowing us to gain that initial confidence. Then comes the data migration. In this scenario, we had to migrate the data because the schema is changing from the Thrift schema to the KV data model. The other such scenarios where we would usually be doing data migration is when we change the underlying database technology itself based on the access patterns. For instance, if we decide that a use case is a better fit on a DynamoDB than a Cassandra, or switching from multi-tenant cluster to a single-tenant cluster. In this particular scenario, we had to migrate the data because the underlying schema was changing from Thrift-based to a KV data model.
Ken Kurzweil: The data migration wasn't easy. We had hundreds of Cassandra clusters and thousands of tables to migrate, and they were all serving traffic live. At the time, we didn't know what we were doing. In the initial launch, the migration already had a system for copying data from one datastore to the other, but it was woefully insufficient. It was an impenetrable black box. We and our stakeholders had no idea where in the data migration process we were, and we knew only whether or not it finished or it failed completely. The existing system lacked safety. If we started transferring data too aggressively and the source and target clusters weren't scaled appropriately, we risked browning out the clusters, impacting availability for applications that depended on them. Even though this issue was critical for the migration of the data, this wasn't critical for the overall migration for KV. We were still able to move KV clusters over and help our customers, specifically, as we were figuring out the data migration.
From their perspective, they were finished. They were happy. The rest was on us. We needed to build confidence in our ability to migrate the underlying data, so we went back to our core tenets. We needed to make it safe. We needed to make it observable. We needed to add validation. To address the safety issues, we retrofitted the existing system to monitor both the source and target clusters and enabling them to throttle back when the key metrics deviated from nominal. We experimented with and implemented various throttling mechanisms. Even simple measures like tracking the p90 read and write latencies proved highly effective. In our approach, it was iterative. We studied the issues as they arose. We experimented with different solutions, and we adopted what worked. The core goal was to enhance system safety. However, as a side effect, the migration process could now utilize spare capacity, scaling up during off-peak hours and scaling down during peak hours.
The safety mechanisms actually made us faster. We enhanced our observability by adding features to report progress during the data transfer. Although initially this was thought of as a difficult problem and it was omitted from the first system, it turned out to be the most effective enhancement to the whole project. Being able to tell a stakeholder that we're X percent done or that this table will take this amount of time to complete was a game changer. This visibility allowed us to identify the clusters that needed to be scaled if we wanted to fit a particular timeline or a budget. Moreover, this new level of insight enabled us to effectively plan. We collected metrics and experience, and we could project the future. Establishing a validatable system was a separate challenge. We started to interview stakeholders to understand their concerns. We implemented mechanisms to catch those discrepancies they predicted. Our parity checking methodology is a separate talk all together. Suffice to say, we worked hard to gain stakeholder confidence.
Once we had confidence in our own migration, in our ability to do the data migration itself, and the dual writes were enabled, all of the data, present to future, was being written to both clusters. Copying the old data from the old cluster, replicating it from present to past to the new cluster was just a process of copying that data. Once migrated, we could parity check the datasets to make sure that they existed on both clusters. Then we can move on to actually traffic switching. Now we can switch and swap reading from the old cluster to reading from the new cluster, effectively changing the source of truth. Once we felt comfortable, we stopped writing to the old cluster, and then we could decommission it. At this point, we were complete. We had migrated. Despite the hurdles and challenges, it was highly successful and we learned a lot in the experience.
Bulk Data Movement
After that migration, we were excited about reusing the technologies that we developed. Things like that throttling for safety. We were looking for places to adapt for more general use cases and data movement in general. We started interviewing application teams. We found that there were application teams that were generating datasets offline, sometimes multiple times a day, into our data warehouse. The team would then move huge amounts of data into their online datastore so that they can serve that data up to their online applications. These things were like signals and metadata for ads, pre-computed rankings for recommendation systems, feature generation based on user activity, and various machine learning applications that were generating embeddings and features. The traditional approach for datasets was simple, but incredibly inefficient at scale, where teams were using batch systems, such as Spark, just to read the data out and write it directly as individual records in to load them.
This often allowed them to effectively DDoS their own applications, overwhelming themselves with their own traffic. We know this as the noisy neighbor problem. The batch data is loaded and consumes resources, introducing contention and impacting the performance of the application. Of course, we could scale, but we end up wasting money as idle capacity when we're not actually doing that batch job. We could apply backpressure, slowing things down, but unfortunately this compromises the accuracy of the data that the application relies on. While we could tune and make tradeoffs for smaller use cases, and we did, they completely collapsed in the face of high order terabyte datasets. At this scale, these approaches took days to load, even though we threw large amounts of money at the problem. This ended up being a critical blocker to high-value business applications that simply could not be served effectively by our existing methods.
When we discussed our use cases with customers who had implemented these solutions, we started seeing a common pattern in their dashboards. They'd have this stack graph with the various versions of their datasets over time. You could actually see in their graph a visualization of the transition from one dataset to the other. We could see it taking hours. During these hours, we could also see that they were actually serving both versions of their data, some of the old and some of the new. Analyzing these use cases, we started to see two categories of access pattern emerge, one mutable and one immutable. The mutable use case, where the application reads and writes from the KV storage system. The batch job in that particular case is really just supplementing the data. In the immutable use case, the batch jobs were solely writing to the datastore, and the application was merely reading from it.
It was really read only. Understanding these access patterns allowed us to simplify a number of use cases and exploit the immutability of this particular set of use cases. These didn't require Cassandra for consensus making. We just needed to serve them read only. We had a theory. Instead of going through the front door and storing the data through Cassandra, we could convert these datasets to RocksDB SSTables and then load these directly onto KV nodes. This should be cheaper and more performant. We don't need to operate a Cassandra cluster in these cases. Simply standing up a KV instance each time we want to do a data evolution seemed like a reasonable approach. The I/O and computationally heavy work of actually doing the file generation could be done entirely offline. We could entirely take out the contention from the online services. We could pre-generate these files on new dedicated KV nodes at line rate as fast as the NICs and disks could allow, and drastically reduce the time it takes to bring up new datasets. We'd stand up new nodes, reroute the traffic, and clean up the previous nodes.
Proof of Concept to Pathfinder
To prove this theory, we built a proof of concept. While there initially was pushback and skepticism about the architecture, the POC was undeniable, it validated our ideas. It worked. It was fast. It was cheap. The POC proved to be the most effective way of converting non-believers to believers. In the POC, the servers implemented the KV API, hosted the data that we generated offline, and had another service that acted as a coordinator that was responsible for relaying and routing information to clients. They could route their requests to the correct nodes. With the functional prototype in hand, we immediately laid out an initial roadmap for the production version. Our initial estimate included building and enhancing the existing infrastructure, which already included self-service provisioning, CI/CD pipelines, deployment systems, but these were stateless systems and we needed to extend them to stateful operations. We were approached by a specific team facing a critical business deadline and theirs was an urgent plea.
We started developing a strategy to accelerate without sacrificing our long-term goals. We looked deeper in our POC and saw that with some changes to the roadmap, we could leverage the existing high-level client and we could build up the POC to allow us to deploy a hardened version of it into production, albeit in a limited way. When we build systems like these, we call them Pathfinders. A Pathfinder is software that you intentionally plan to migrate from. A Pathfinder unblocks business-critical use cases for stakeholders willing to take some risk, and allows you to learn as much as you can while providing value. A Pathfinder is a way to have a symbiotic relationship with your stakeholder. They get a feature they wouldn't get otherwise and you get to discover the unknown unknowns with a willing participant. We abstracted the POC client's implementation of the name, resolution, and routing, and we put that behind a feature flag and released an improved high-level client for KV that enabled us to seamlessly switch between the existing KV feature set, the Pathfinder, and to the yet-to-be-delivered production KV implementation. This meant that an application team wouldn't have to change their code when our actual product was ready, and we could transparently switch their traffic from the temporary setup to the production infrastructure.
The Pathfinder was fast. Fundamentally, it was simple and we understood it. We built tools and methods for benchmarking the amount of time it took to deploy datasets, and we well understood the performance regarding throughput and latency. We used these benchmarks to compare the performance of the KV production implementation as it was being developed. They were not equal. The KV team was able to use our benchmarks and run them themselves and focus on the bottlenecks and renewed them, repeating as necessary until we were all confident. Eventually producing a version of KV that was not only comparable to the performance of the Pathfinder, but the improvements were generally applicable to all KV use cases and benefited all application teams, regardless of this new tech we were building. The Pathfinder also allowed us to learn how to convert datasets and store them, and stage them in S3. Once they were in S3, we could deploy them at will.
As soon as we finished the high-level enhancements, we could deploy to production and unblock critical business need. We deploy a new version of the coordinator, hardened for production. We get the application team to pull in the newly released client. Our control plane is then instructed to download the dataset and update the coordinator, and the clients read from the coordinator and route requests. This goes on until another dataset is prepared offline and staged in S3. The new nodes are brought into existence. The data is loaded, the coordinator is updated, and the previous nodes are removed.
KV Stateless to Stateful
Rajasekhar Ummadisetty: With Pathfinder, we were able to unblock and support the critical application team. This not only addressed an immediate need, but also gave us valuable time to build towards our production infrastructure. The learning from the Pathfinder set the foundation for this work. As previously highlighted, this required transforming our key-value abstraction from a stateless system into a stateful one, and updating our deployment process to not just handle the code deploys, but also data deploys. To recall the key-value abstraction stateless architecture, at a high level, we had servers that were stateless. They connected to the control plane, fetched namespace information, connected to the database, and started up. Once all the servers are started up, applications are able to now send requests to it. The request can be handled by any one of these servers because they are all identical and stateless. They just transform this request into a database query, execute it, and respond back to the application.
With the move to the stateful architecture, the servers are no longer stateless. Each server stores a subset of data, making every server unique in terms of the data it manages. To coordinate this, we introduced several new control components. First, there's a partition group manager. The server communicates with the partition group manager to acquire a unique lease, which defines the specific subset of data the server is responsible for. We also introduced a central routing manager. The routing manager keeps track of all the KV clusters that are stateful, the namespaces that are stateful, and within each of those namespaces, which server hosts which part of the data. Even within a KV cluster, we introduced a cluster routing layer that talks to the central routing manager and caches information specific to that KV cluster. It tracks like what namespaces are available within that KV cluster and where exactly is the data hosted.
The additional responsibility of KV cluster, cluster routing, is to relay this routing information to the application that connects to this cluster. The client on the application can then use this routing information to route requests to the specific data nodes. We also extended our deployment infrastructure to support the stateful data deploys. When a new data deploy is triggered, the first step is to capture the desires. These are the requirements from the application team, such as, what are the expected reads per second? What is the dataset size? We then transform these application-level desires into infrastructure level desires, such as, what type of instances to use, how many instances, how many replicas, what is the dataset to be hosted? Based on desires, we then provision an autoscaling group and then wait for the nodes to come up. Each server node first talks to the control, gets the namespace, similar to the stateless architecture, but then it talks to the partition group manager, gets a unique lease, as previously mentioned, to determine what part of the data it needs to host. Then goes to the S3 and fetches the subset of the data files that were generated offline, and pulls those files, loads them and bootstrap. Once all the nodes are up, we then enable a staging stage. The staging mode is for the application teams to validate their dataset before they promote it to production.
If you recall, the traditional approach to data ingestion resulted in what we call the sawtooth pattern. In this model, the older dataset in the datastore was gradually replaced by the newer dataset over a period of hours. During this slow transition, the application teams had a window of time to observe the gradual change while the data was being replaced. This allowed them to set up monitoring and alerts to catch any regressions or data corruptions as they happen. Although a dedicated validation system would have been ideal, teams were able to work around this limitation by relying on this lengthy ingestion window to monitor and respond to issues. With our updated process, this transition window becomes much smaller, almost instantaneous. The data ingestion itself may take a few minutes, but once the new data is ingested into the new nodes, the switch from old to new dataset happens in a matter of seconds.
As a result, the previous monitoring and alerting mechanisms are no longer effective, because the application is suddenly exposed to this entire new dataset all at once. This increases the blast radius in case of any issues such as data corruption, since there is no longer a gradual rollout. Because of this, the validation becomes even more critical before we switch over the dataset from the old to the new. That's one of the key reasons why we introduced a dedicated validation stage into our new deployment process. In the staging phase, the application team can spin up a validator application, and that allows them to connect to the staging cluster, execute requests, validate the data, build the confidence, and then give us the signal to promote the dataset. Once we receive this signal, we update the routing. The routing information gets communicated all the way to the application. The client then uses this updated information to route requests to the new dataset.
We wait and monitor to see that all the requests get drained from the older dataset to the new dataset in a matter of seconds, but we still keep the older dataset around in case we see any regressions or issues, we can instantaneously roll back. Once we have the confidence that the new dataset is working as expected, we then destroy the older dataset.
Pathfinder to Production
With that, we completed the transformation of our key-value abstraction from a stateless to a stateful infrastructure, and our production infrastructure was ready to support critical use cases. If you recall, we unblocked one of the critical applications using the Pathfinder infrastructure and it is now time to migrate that critical application from the Pathfinder infrastructure to the key-value infrastructure, and also validate our assumption that abstracting away and putting our initial focus on enhancing the client would give us this flexibility of migrating this transparent to the customer application team. We executed this migration in two carefully controlled phases to limit the blast radius and ensure a smooth transition. In the first phase, we replaced the Pathfinder coordinator with the KV cluster routing. Importantly, the KV cluster routing continued to relay the routing information to the application so that the requests are still directed to the Pathfinder data nodes.
This allowed us to incrementally introduce the new component while minimizing the risk. It was also a reversible two-way door. If any issues arose, we could quickly switch back to the original coordinator. Once we gained the confidence the application traffic was being correctly routed through the KV cluster routing, we proceeded to phase two. In this phase, we loaded our KV data nodes with the same dataset that was present on the Pathfinder nodes. Once the KV nodes were fully loaded, we updated the routing information so that the client requests would be now directed to the KV data nodes instead of the Pathfinder nodes. As requests transition to the KV nodes from the Pathfinder nodes, we complete the migration. This migration was entirely transparent to the client application team. No changes or actions were required on their end, and all configurations were managed by us safely. If we observed any regressions, we could have instantly reverted to the previous state. Throughout this process, we relied on our safety, observability, and validation measure. The benchmarks we ran beforehand gave us that confidence that the KV production infrastructure would perform on par with the Pathfinder infrastructure. In fact, the application team did not notice any difference.
Wins
Our commitment to innovation and cross-functional collaboration delivered not just a technical solution, but a massive business value. I want to highlight three critical use cases that were previously struggling with our traditional data movement process and how we enabled their success. The first use case involved moving massive datasets. With the traditional approach, it was simply impossible to meet the customer application's SLO requirements. The load times would have taken days, incurring infrastructure costs in millions annually. With the new architecture, we now can load dataset in under 40 minutes and switch it over in seconds. This translated directly into 99% reduction in the deployment time and 70% in cost savings. The other use case involved high-velocity versioned feature data, smaller in size but extremely higher in frequency. We reduced the data load times to less than 10 minutes and achieved operational savings of 90%. Additionally, this use case required the ability to roll back and roll forward between dataset versions, a capability that was previously unavailable.
The third use case, perhaps one of the most powerful wins, was enabling platform development. One application team was able to build a robust machine learning platform on top of our solution. This platform now empowers multiple teams to rapidly deploy and consume custom ML features, proving that our work wasn't just a technical fix but a true accelerator for the business innovation. These critical use cases unlocked major business investments and led us to formalize the architectural principles that made all of this possible.
Generalizing the Architecture - Capture, Conversion, Deployment
Ken Kurzweil: While developing the system for bulk data movement, we generalized a framework that we call capture, conversion, deployment. This framework enables us to build systems in a modular, reusable, and repeatable way. This allowed us to divide the work amongst teams, mainly because the inputs and the outputs of each of the phases are well understood by each of the teams. Traditional methods would have resulted in a single coupled system that would take a dataset and deploy it. Splitting the phases into capture, conversion, deployment allowed each of the phases to be decoupled and developed and optimized independently. In the capture phase, we create and store an immutable artifact of the dataset, ensuring that it can be referenced and provide us the original state of the dataset at the time that we captured it. The primary goal is to provide an unalterable artifact for later processing. Capturing can either move directly into conversion, or we can use several captures to actually create deltas, generating a more efficient deployment artifact.
For KV, in the conversion phase, we can convert the capture artifact into the format that is designed and optimized for deployment. For KV, we used RocksDB, but this could be any format. The objective is to tailor the dataset to the deployment artifact so that it's optimized for each target. The deployment involves orchestrating and transferring the stage artifacts to the datastore and bringing those datasets online in a deterministic manner. This could include provisioning a new cluster, loading the data, or managing the traffic switching. When you do the deployment step, it is independent of the capture and conversion steps, and this is critical because it means that we can roll forward or backwards to the previous datasets in a straightforward way. Our application teams are updating their dataset all the time continuously. Our system captures that dataset, either triggered or scheduled, and then the capture converts that into the deployment format.
In our case, such as RocksDB SST files, as previously discussed, or most recently, Cassandra SST files, which are currently under testing. This capture, conversion, deployment methodology can be extended and provides us a way of being able to reason about future enhancements and optimizations. By breaking the data movement into three discrete steps, we can reuse and interchange components, and we can integrate new conversion phases for new datastores without disrupting existing processes. When one component is optimized, say, a faster method of dataset capture, the performance benefit is extended across the ecosystem. The framework empowers infrastructure and application teams, making it easier to deploy and roll back their dataset independently. This ability facilitates data validation, performance testing, and mitigation. Deterministic artifact deployment can be used as a reliable safety net and enhance confidence and safety during these data movement operations. The distinct phases allow teams to collaborate, and in our case, it was the data movement team, it was the KV team, it was the UX team. They can all work independently on different components because they can validate each of their own deliverables.
Future
Let's talk about the near future. We're actively working on taking the learnings that we've most recently learned with immutable datasets and applying them to mutable datasets. In our particular case, the capture phase would stay the same, but the conversion phase now can produce Cassandra SSTables using the Apache Cassandra analytics project. The deployment phase includes loading and generating these files into a live Cassandra cluster actively serving traffic. Our custom orchestration and Cassandra's built in file loading and compaction and conflict resolution mechanisms allow us to move forward. We've built this and we're actively testing it, and we're excited to share in the future. I'm excited about the ecosystem of data movement tools and the techniques that we now have, and here's a glimpse of the future. We've revolutionized our ability to read and write Cassandra SSTables from and to Cassandra clusters. All the work that was done previously applies here, but we have new capabilities.
We're already able to use mechanisms to export Cassandra SSTables into S3. This is how we do backups. We've recently upgraded our ability to read and convert them. Even more recently, we gained the ability to write these, import them into live Cassandra clusters. Furthermore, the capability exists for us to write to any datastore format and allow us to migrate to any datastore behind our abstraction layer. For example, switching from Cassandra to DynamoDB if the cost or performance tradeoffs warrant it.
Key Takeaways
Here are the takeaways that I hope you can carry back to your organization. Confidence is currency. Safety, observability, and validation are a set of architectural principles that we consult each time we need to gain confidence. Abstraction is a superpower. Both abstraction and our ability to use indirection allow us to perform large yet seamless migrations. You should analyze and exploit the access patterns, talk to your stakeholders, understand their use cases. That's where innovation is found. A POC is worth 100 meetings. We've found it crucial in convincing stakeholders. Pathfinders allow a way of moving faster by being intentional about the things that you want to learn, how you want to migrate, and what are you going to throw away? POCs and Pathfinders help us plan the future. Finally, leverage architectural patterns like we did with capture, conversion, deployment. This allows teams to think about systems and components built faster and independently.
Questions and Answers
Participant 1: During the entire migration, what were the key observability metrics that helped you make sure that the application teams still have the correctness and idempotency they expect on their end?
Ken Kurzweil: Generally, the key metrics that we're looking at are throughput and latency and error rate. Beyond that, our customers were integrated in the process of doing the migrations, so they had heads-up as they were transitioning over. Most of it relied on us in being able to provide them guarantees that we already did the parity check, that most of this was behind us as far as being able to have that confidence. The thing that we were most fearful of is missing a few records, not millions of them. Most of the problems you wouldn't see for months if there was an issue, so it needed to be zero.
Rajasekhar Ummadisetty: For the initial migration, for the Thrift-related migration, that was one of the advantages of having this abstraction layer is you can build in idempotency into your abstraction. All of our mutating APIs had idempotency tokens. That guarantees that all these requests, whether going to the different backends, they still maintain that idempotency.
Ken Kurzweil: Why did we use RocksDB? We actually looked at a number of different file formats. Initially, we looked at Hollow. That's something that's developed internally. It's a pretty cool piece of kit. At the time, and still today, it's something that fits in memory in JVM. It was an excellent use case. We also looked at, specifically, SQLite. SQLite is an excellent file format. I'll tell you why we didn't choose that. We looked at Cassandra SSTables first. We thought that we could certainly be able to write them, but they were less supported as a separate entity, as a format by itself. As we were shopping this around with our believers and nonbelievers, one of the things was, why not use the ext format for Memcached? Out of all of those things, what we had was a matrix that basically showed us, should we look at how well is it supported and documented?
Is it a format that's actually designed to be consumed externally? What exactly does it do for compaction? If we have two files, does it automatically compact, or do we have to compact it for you, and other metrics. For example, SQLite fit all of the bill, except it doesn't compact. You can dynamically load it. You can basically open the SQLite library. You can actually connect two different databases. You can't tell it to merge two tables. With RocksDB, this checked all of the boxes. It was well supported. It is an external format that is designed to be used that way. It had online compaction and the ability to mount.
Participant 2: In the same grain of RocksDB is that, did you look at ClickHouse which also does a very similar form of compaction, and all this?
Ken Kurzweil: No, we didn't look at ClickHouse. Is it specifically the file format, is a separate file format?
Participant 2: The same file format, time-series data.
Ken Kurzweil: No, we didn't.
Participant 3: While you were trying to figure out which clients that you could convince to start using as part of Pathfinder, did you have a strategy in mind to go with, shall I take the hardest problem first, where the biggest client or the biggest use case is, let me go after them and try to convince them? Or did you take an approach of, let's find a client who has new feature, which is not business critical, even if it gets delayed or it gets any of the reliability issues or data fidelity issues, it is still an ok thing to be, because Pathfinder is still something that both you are trying to prove for yourself as well as for the customers.
Ken Kurzweil: For us, in that Pathfinder phase, you're really looking for early adopters that want to have skin in the game. They're looking for a new feature. For us, we were specifically looking for something that would extend to the rest of the team, but there were specifically candidates that we found that it was a deal breaker if we couldn't deliver something. They really needed the product. The specific customer that we're talking about in this one as the use case for us to go to the Pathfinder, it was a very large use case. It was very much mission critical and it was hinged on other pieces. They had very much an appetite to take a little bit of risk and help us through the journey as a particular customer. As we're picking out specifically in that Pathfinder phase, you want to find stakeholders that are willing to work with you, with those unknown unknowns, as you're going through the process.
If you can find those, those are excellent, because if something goes a little bit awry, they're going to be the first ones to tell you. They're going to be your best customers. Specifically, in this case, they're one of the largest use cases that we actually had to build for, for this. It was kind of, I don't know if you've heard the cliche, eat the frog, when if you eat the frog at the beginning of the day, you don't have to eat it the rest of the day. This was very much, eat the frog. If we could solve this particular problem, we would most definitely be able to take care of the rest of Netflix. Once we had a comfortable implementation at that point, I felt confident that I could move forward. Absolutely, you want to find the right customers. Then, specifically, you want to find the ones that are going to stretch your implementations so that when you go into the field, you didn't do a bunch of very simple things and go to production and find that the general use case is several orders of magnitude different than what you expected.
Participant 4: When you moved from front-loading into Cassandra to RocksDB SST, can you share what was the load time savings that you had if you went directly into Cassandra to back-loading into RocksDB?
Ken Kurzweil: It was tremendous. In our particular case, the standard deployment schedule for our largest customer that we were looking at was that it was taking several days to actually load a multimillion-dollar cluster with a reasonable amount of speed. That still wasn't good enough for this particular customer because they wanted to evolve their data at a higher rate than the several days that it took us to actually load that data. The Pathfinder specifically, in its implementation, we got that down to 27 minutes. It was a complete success from a standpoint of proving us this idea. As we rolled into and added more of the hardening, that number went up to 40 minutes, but that specifically told us that this is something that is a valid way to go. We're currently seeing the same kind of impact as we're moving towards Cassandra, but just not as equivalent, just because as we're loading SSTables for Cassandra, you have to contend with actually merging state. You're actually doing that compaction phase, wherein what we just described here, the entire thing is happening offline and we're swapping new instances in place. It's very quick.
Participant 5: Did your abstraction layer add latency or cause an issue?
Ken Kurzweil: It certainly does add a layer.
Rajasekhar Ummadisetty: Yes, it does. What we observed is the additional latency that it adds is not so important for our application teams versus the benefits it provides on top of it. We initially had that fear that it would be harder for our application teams to convince that, ok, we are going to add an additional hop. Once we showed the benefits of this, so the tradeoff was much better.
Participant 6: Somewhat relatedly, you talked at the top of the talk that you had many different other data abstraction layers in addition to the one that we talked about here. I was curious if you ever encountered a case where you tried to develop an abstraction layer and it turned out to be the wrong tool for the situation, and what happened there?
Rajasekhar Ummadisetty: We did. There are cases where the abstraction cannot be generalized into this subset of the APIs. For instance, if you take relational DB, you have the SQL interface, which is like a wide API surface area. If you want to abstract SQL API using some subset of APIs, you might not succeed to solve all the use cases. Those would be the cases where you have to really think like, how can you narrow down the API surface area and really fit into our abstraction?
See more presentations with transcripts
