So my name is David Shepherd. This is Jonathan Indig. Um, we are software engineers at Netflix, and today we're going to talk about a system that we work on built on Clojure and Datomic.

Uh, and we hope to give you some insight into how this system makes use of Datomic in particular and how it has empowered the project's success over the last decade.

So, uh, we'll talk a little bit about kind of what it is, um, what we built, how it's evolved over the last 10 years, uh, and then Jonathan's going to talk about how stability has been important to us as an organization and contrast that um, with how instability uh, affects us.

So, what do we build? Um, so this is not actually the first time we've introduced the system.

Our dear friend and colleague Matt Bostonbrook gave a talk at the Conj almost 10 years ago to the day uh, introducing this system.

Uh, Do or Do Not, There Is No Try. Uh, this was a system for uh, uh, for building data pipelines.

Um, and if you've not seen this talk, I highly recommend you go watch it. Many of the core ideas that went into how we think about data pipelines, um, how we model them, are are still very much alive and well today.

We've made many advancements over the last 10 years, but this the still core ideas are still true um, to this day.

Um, so what is it? So, uh, it's for building data pipelines.

Um, and it's it's really for data scientists and researchers to define uh, data, in particular ML uh, ML training pipelines.

Um, so we give them a simple Python API. No tomatoes, please. We got to meet them where they're at.

Um, but it essentially allows them to define these pure Python functions. They can decorate them. They compose these functions together.

Uh, and we essentially take that call graph on the right and we translate it into EDN. We turn the date math into Clojure expressions that we can evaluate in SCI and that resultant payload is basically what we use to compute the underlying um the underlying DAG.

So we can take that example and run those, right?

These run in distributed uh in a distributed way on separate um instances. Um, now that's all well and good. Many orchestration systems can do this.

It's not particularly interesting, but um the the the feature that sort of sets Dagobah apart is that we have a very um robust model to track data provenance, right?

So we know the code and data that flows through every single one of these steps.

And so we can actually do structural sharing uh like you you would do with persistent like Clojure does with persistent data structures.

We do that for um data pipelines. So in the example above, you can make a structural change to the pipeline to sort of edit let's say the depth in which you wanted to recurse.

Uh and we can work out the part of the pipeline that actually we can totally reuse and skip and memoize and we only have to compute the necessary delta to materialize the sort of new result that this researcher wants.

Um, needless to say this is a huge productivity boost for researchers in Netflix.

Um, two researchers working on the same pipeline will automatically share the common components.

They're not restarting from scratch. You know, these jobs, you know, they're not trivial things doing simple math in Python.

These are like Spark jobs processing hundreds of terabytes of data, taking hours or days to run. Um, and so not having to do them is a big win, not just for time, but money saved for Netflix.

So, um, this system is currently responsible for orchestrating all personalization ML data prep and model training at Netflix.

Um, uh, we have no dedicated Datomic or Clojure support team. Uh, I mean, you're you're looking at them, basically.

Um, we we do, however, pay for an enterprise Datomic support contract. Um, this is extremely critical infrastructure for Netflix and we are we do not want to go it alone, right?

We take all the resources we can, uh, and we have fantastic partners, um, helping us do it.

So, in the beginning, um, Dagob was built to support a single ML team, right, from its genesis, right? Um, so early days, the architecture was quite simple, right?

It was deployed essentially as a monolith in an HA way using Zookeeper to sort of manage leadership state. Um, and this one one deployment would manage all aspects of the application: handling API requests, launching jobs on underlying compute systems, updating state and persisting that into um, and we deployed Datomic on-prem backed by a Cassandra cluster, um, with a sort of shared Memcached deployment as well.

Um, now there's a couple interesting properties of the the state machine that we maintain. Um, it's actually written as a forward inference rules engine, not too different from like Clara rules.

Um, our rules are structured around essentially a Datomic query to a pure function that produces transaction data.

We run those rules with the result onto a database and run that recursively until we reach a new stable state and say, given this novelty here, here's what we must persist, and we write that down.

Um, now these rules query against what we consider currently active work. So we need to look at what is currently transient in state.

So we don't want to be considering, you know, finished executions from days or weeks ago.

We just want to be looking at the current state of now. And the way we do that is we actually aggressively retract inactive entities for performance.

So our uh asserted true database is basically what is currently running and what is retracted is sort of done in the past per se.

So most API queries to look at history are actually looking at a history data database in Datomic. Um, an example of one of these rules might be, find all newly failed jobs and cancel their dependents.

Um, now needless to say, the system has evolved a bit in the last 10 years.

Um, that architecture worked well, right, to understand how to model the domain, um, you know, and to serve that one team, but now it's used by dozens of teams training thousands of models.

Uh, and Netflix as a business has changed a lot in the last 10 years and so we've needed to evolve the application to keep up with these sort of evolving needs.

Um, and you know, as with most monoliths, we had a lot of new heterogeneous processes, right, all contending for resources in a single VM.

Um, several of these benefited from horizontal scaling, um, and so we knew we needed to start splitting the system into smaller deployable units to sort of, you know, access more compute and to also improve sort of overall efficiency of the system.

Um, but we still wanted to share the same codebase and component system.

I think Netflix is known for going microservices, but we were a small team. We didn't need to go full-on to have totally independent teams of microservices.

We wanted one shared codebase, and we just needed to be able to deploy the app in different what we called stacks.

So the first piece we wanted to tackle was the easiest, right? We needed to sort of build our muscle in understanding how to sort of split things out, um, in this deployment architecture with Datomic.

Um, and so we didn't want to have to solve additional sort of distributed systems problems.

We just wanted to understand the mechanics of doing the separation and understand whether our understanding of the performance we would get was actually true.

Um, so the first one we tackled was a component called the reporter. Um, its job is basically to run long historical queries to emit information to a time series database central to Netflix.

We use it for like dashboarding and alerting. Um, and it was basically running these sort of standard queries on a regular period.

But these queries were creating increasing CPU pressure as the usage was going up. Uh, it was also reducing cache coherence, right?

So you have these rules running against things that have happened seconds or hours ago competing with cache space from things that are looking from a week ago.

And so, uh, it was a relatively simple split, right? We deployed it in a similar HA way.

Um, and it basically had no need to talk back to the direct application, right? So it was a fairly straightforward split uh, and it could function as a standalone peer, uh, and and query Datomic directly.

Um, but we needed to make sure that read load was decoupled from write load.

This was the fundamental assumption we wanted to verify. Right? With most databases you need to spin up like read replicas, otherwise you're still competing with resources on your kind of primary write master.

Um, but because these queries are so regular and the peer has a built-in object cache, the cache miss rate on this is 0.0013%.

So virtually no queries are actually touching head-on storage, right?

So this effectively functions as a total island and that we can scale totally independently from the rest of our application.

We just don't have to think about it, right? Which is fantastic, right? It was a very straightforward approach, and we got exactly the result we had hoped for.

Um, so that gave us confidence to sort of move on to the next portion that we wanted to actually horizontally scale, and the obvious thing there is like all your read APIs, right?

These are all stateless queries. They don't update any state.

We can serve those and scale as needed, right? So the API component as it were. Um, but you know, it wasn't as regular.

It wasn't homogeneous queries. There's lots of heterogeneous things. Users want to see their graphs that are currently running.

They want to run timeline queries to understand the state of things from weeks or months ago. Uh, and so it wasn't obvious that necessarily we'd get the same sort of uh performance trade-off that we would with splitting out the reporter.

Um, but it was still had these same problems, right?

We were getting CPU pressure and memory cache coherence problems for the exact same reasons. Um, so here, right, we had to spin up a horizontally scalable pool of API instances, but now we had to mediate our read and write traffic, right?

So we would still forward writes to our main application and then just handle the reads directly from within the API instances.

And so in this case, right, we still need to verify, okay, is our read still truly decoupled from our write load because these queries look very different than a very standard, fixed query running on a cadence.

And even in this scenario with different queries from all the different customers spread across all these instances, the cache miss rate is still 0.02%, right?

And so again, we're in a scenario where we can scale this horizontally and we effectively see no—we can totally isolate or buffer the load that would otherwise hit the shared resource, almost like in the power strip, um, that's shared with the critical write path.

But we had a new problem to solve, which is now read-after-write consistency, right?

Because reads are now happening on separate instances from where writes are happening. And our clients, we don't want to expose this internal detail and say, hey, you write this thing and you might not see the result right there's a slight lag between a write goes through and every peer sees it.

But Datomic has two lovely primitives to address this problem. The first is the basis T value you get from transacting.

something it's just a simple number, so it's easy to pass around in an HTTP header or in the response body. Uh, and then you can call `desync`, right?

You pass this T value in on any peer you want, and you are guaranteed to get a database at least up to date with that value.

So we can just synchronize the API instance that forwarded it. It can run all its resolvers against the state of the world that's guaranteed to have seen that write.

And we can forward the basis down to the client. So all subsequent reads to different peers will still see the result of that write.

Uh, and doing this is basically just a plumbing job, right? It's not like a hard distributed system.

We just have to pass the information around and it just works. Um, so the next scaling challenge was a little bit more interesting and this had to do, I had mentioned that we have to aggressively retract things that are no longer active and we had to do this in a sort of an atomic way.

As soon as a graph is inactive, we would atomically retract the whole piece. Um, now these transactions were occasionally getting kind of large.

Um, and some detail on this, our graphs are actually modeled as nested component entities. So you'd have like a root, it would have component entity relationships with its dependencies and that would sort of occur recursively.

Uh, and this was convenient because you could simply retract the root and the whole sort of connected graph would get retracted with it.

Um, but we were seeing that retracting these single graphs in one shot was starting to occupy a transactor for multiple seconds.

Um, and the reason is these graphs were getting a little bit big.

Um, so this here is a representation of the DAG necessary to compute one uh one retraining of a messaging model. This is a model that predicts um uh the best sort of message content, whether or not to send it on your your iOS device or as an email or something like this.

Um, except this isn't the whole thing. Um, I had to sort of do some collapsing to fit it on the slide.

It actually looks something like this. And we were trying to retract multiple of these in a single transaction because a shared upstream failure might affect multiple of these in one shot and we were trying to retract it all at once.

Needless to say, this is not a great thing on any OLAP database to transact something that large, but we were doing it.

Um, and we needed to fix it because those long long latencies were not great, right?

We needed to sort of address that. Um, so we did as we did before.

Um, we split up this archiver onto its own stack so we could isolate the sort of resources it needed. Um, but we also needed to move to tombstones, right?

Instead of retracting directly. Um, because when you retract in Datomic, you're actually retracting all of its attributes sort of separately, right?

Um, and so instead of retracting the entity, we just mark a tombstone boolean.

Um, but the beauty of Datomic is schema is very flexible. So it's very easy to just add this new attribute and all the different entities in which we need to do this.

We can just tack it on. Oh, and we can index it too.

Um, but now we need to sort of exclude these tombstones. Whereas before these entities were sort of um, excluded because they were no longer true.

Um, they were true but had this separate bit that we needed to sort of exclude.

A Decade on Datomic - Davis Shepherd & Jonathan Indig (Netflix)

Clojure (https://clojure.org) is a robust, practical, and fast programming language with a set of useful features that together form a simple, coherent, and powerful tool.

ClojureTV brings together talks and presentations from the community, major Clojure conferences, and Cognitect (https://cognitect.com).  Here you can find Clojure/conj, Clojure/west and EuroClojure talks, as well as custom video tutorials and presentations from Rich Hickey and the Cognitect team.

A Decade on Datomic - Davis Shepherd & Jonathan Indig (Netflix)

Scripts