Tying it all together: build a social network

The depots correspond to the categories of events happening in the application. Not every data type has its own depot. Some of these depots receive multiple data types. For example, the "*friendRequestsDepot" receives both FriendRequest and CancelFriendRequest data. This is necessary so that different types of data that affect the same PStates are processed in the order in which they happened. As an illustration of this, suppose a user is spamming the "request friend" and "cancel friend request" buttons on the RamaSpace UI. This will generate many FriendRequest and CancelFriendRequest records. If those were kept on separate depots, there’s no guarantee as to the order in which they will be processed. The user could generate data in this order:

But they could be processed in this order:

Only by keeping these two datatypes on the same depot can the order of processing match the order in which the data was generated.

Likewise, how a depot is partitioned is critical for ensuring this ordering. The order in which data on different partitions of a depot is processed is undefined because they are being processed in parallel. If "*friendRequestsDepot" were defined with Depot.random() partitioning, you will run into the same ordering mismatch. So "*friendRequestsDepot" instead defines its partitioning as Depot.hashBy(UserIdExtract.class). This class ensures FriendRequest and CancelFriendRequest from the same user always goes to the same partition. UserIdExtract is defined as follows:

This uses the helper class TopologyUtils.ExtractJavaField from rama-helpers to extract the userId field from the data being appended. Both FriendRequest and CancelFriendRequest have a field userId.

The definitions of the other depots all use the exact same logic. You’ll see specifically how the data from each depot is used in the definitions of the ETL topologies.

The users topology handles user registrations and profiles. Let’s start with the declaration of the topology:

Streaming is used for this topology because low latency updates (a few milliseconds) are necessary for these use cases. Next is the declaration of the one PState for this topology:

This is the "fixed keys schema" that was mentioned before. For each user ID, a map with a fixed set of fields is stored. These fields represent all profile information for that user, and each field has its own schema.

Next is handling user registrations. The key task to handle is detecting whether the registration succeeded or whether the requested user ID already exists. An incorrect way to do this would be to first query the "$$profiles" PState to check if the user ID exists and then append a UserRegistration if the user ID is available. This doesn’t work because there’s a race condition between two clients trying to register the same user ID at the same time: they could both see the user ID doesn’t currently exist and then both append a UserRegistration record.

The correct way to handle this is within the ETL code that processes UserRegistration records. Each UserRegistration is an attempt to register that user ID – if the user ID currently exists, it doesn’t do anything. In order for a client to know whether their registration succeeded or not, the client provides a UUID with the registration that gets written into the "$$profiles" PState if and only if the user ID didn’t exist. Here’s the code:

This code takes advantage of fact that events for a single task run in sequence. So while this code is processing a UserRegistration, no other events for the same user ID can be running at the same time (but events for different user IDs on different tasks can be running in parallel). This is a property you can take advantage for other goals, such as updating multiple PStates on the same task and ensuring those changes become visible atomically.

This ETL code takes advantage of the depot partitioning for "*userRegistrationsDepot". The depot is configured to partition by the user ID inside the UserRegistration objects. So when this ETL code begins processing a UserRegistration, it’s already on the task with the correct partition of the "$$profiles" PState for that user ID.

The next step is extracting the information needed from the UserRegistration to perform the work of the ETL. Here you can see a facility of Rama’s dataflow API that hasn’t been introduced yet: macros. Macros are a way to decompose and reuse chunks of dataflow code. macro is given a block of code and expands that block of code into that callsite. This macro call uses the extractJavaFields function defined in rama-helpers and expands to:

ExtractJavaField is also defined in rama-helpers and extracts the public field from a Java object with that name. The extractJavaFields function produces code which maps the desired output variable names to field names. There’s much more to say about how to use macros, and you can read more about them on this page.

The rest of the ETL code is straightforward. It writes the user registration information into the $$profiles PState only if that user ID doesn’t currently exist. The key code is the filterPred(Ops.IS_NULL) navigator, which only continues navigation if the user ID doesn’t exist.

With the ETL explained, we can now complete the story of how user registration works. A client does the following procedure to attempt to register a user ID:

Here’s the client code for this:

Although only a small amount of code is required for handling user registrations, both in the module and client-side, it’s a great example of composing together Rama’s primitives to achieve application-level goals.

The last piece of the users topology is handling edits to existing profiles. This is accomplished by processing ProfileEdit data from the "*profileEditsDepot" like so:

Like user registrations, this code takes advantage of the partitioning of "*profileEditsDepot". It simply reads the field and value from the ProfileEdit and overwrites those in the "$$profiles" PState.

The friends topology handles friend requests and maintaining a bidirectional social graph. A user can view both incoming and outgoing friend requests, can cancel pending friend requests, and can unfriend existing friends. Let’s start with the declaration of the topology and the PStates:

This is a use case where either streaming or microbatching could be used depending on what exactly the application designer desires. Streaming ensures new friend requests and friendships are visible very quickly, while microbatching would be able to handle higher throughput while having higher latency (on order of a few hundred milliseconds).

There are three PStates for this topology all with the same structure: a map of user ID to a subindexed set of user IDs. These PStates track outgoing friend requests, incoming friend requests, and friendship relations. Any friendship will correspond to two pieces of data in the "$$friends" PState for each direction of the friendship. Each direction of a friendship could be stored on different partitions of the PState since the two user IDs are unrelated.

Let’s now take a look at the code for handling friend requests and cancellations of friend requests:

There’s a new operation here that hasn’t been introduced yet: subSource. subSource dispatches processing based on the type of the input data. While you could accomplish the same thing here using ifTrue, subSource is especially convenient when you have more than two data types. In this case there are two branches of computation: one for the FriendRequest type, and one for the CancelFriendRequest type.

Let’s start with the processing for FriendRequest. "*friendRequestsDepot" is partitioned by the userId field, so the processing is already on the right task to update "$$outgoingFriendRequests". This is done here with compoundAgg by adding "*toUserId" to the set of outgoing friend requests for "*userId". You could also perform this write with localTransform, but the code is slightly shorter with compoundAgg. Next, the computation is relocated to the partition representing data for "*toUserId" and a similar compoundAgg call is done to update incoming friend requests for "*toUserId".

The code handling CancelFriendRequest is the same, except removing elements from those nested sets instead of adding them.

Next, let’s take a look for handling friendship additions and removals:

The idea here is when a user accepts a friendship request, only a single FriendshipAdd record is appended to the "*friendshipChangesDepot". So this code is a little more involved because friendship requests must be cleared in addition to updating the "$$friends" PState.

This code uses "anchor" and "hook" to create two branches of computation. The first branch clears all incoming and outgoing requests between those two user IDs, and the other branch updates the "$$friends" PState in both directions.

For the posts topology, let’s once again start with the declaration of the topology and the PStates:

Because this uses a microbatch topology, it will be able to handle very high throughput at the tradeoff of moderately increased latency of processing. Microbatching is used here for the purposes of illustration; whether you use a stream topology or microbatch topology depends on the latency you desire of new posts being visible on the user’s wall.

The "$$posts" PState is a map from user ID to post ID to Post. The inner map is subindexed because users will have lots of posts on their wall. Post IDs are generated in descending order, which causes a user’s wall to be stored in reverse chronological order. This is exactly how we desire posts to be ordered on the frontend. Getting a page of posts is as simple as using the sortedMapRangeFrom navigator, as you’ll see later in the resolvePosts query topology.

TaskUniqueIdPState is a utility you haven’t seen before that comes from the rama-helpers repository. It’s a simple tool that generates IDs unique to this task. It keeps the next post ID value in the local partition for the "$$postId" PState. So the "$$postId" PState is partitioned numbers – not a map like you’ve been seeing so far. In this case the TaskUniqueIdPState is configured to generate IDs in descending order starting from the largest Long value.

TaskUniqueIdPState generates IDs that are unique on the task. These IDs will not be unique across tasks. In this case that’s all that’s needed. rama-helpers contains another utility called ModuleUniqueIdPState which generates IDs unique across the entire module. It does this by incorporating the task ID in the generated ID.

Now let’s look at the topology code:

This is only five lines of code! The only new bit is the genId macro from TaskUniqueIdPState which generates a new ID and binds it to the "*id" var. Helper utilities like TaskUniqueIdPState often expose fine-grained functionality like this through macros.

Also of note is the lack of any partitioner calls: the partitioning of depots is respected by microbatch topologies the same as stream topologies, so a post begins processing here on the same depot partition it was appended to (partitioned by toUserId).

The last topology in RamaSpace is the profile views analytics topology. Recall that this ETL produces a PState to support a query for getting the number of profile views over a range of hours. Once again, let’s start with the declaration of the topology and PStates:

Microbatching is perfect for this use case. It’s completely fine for the update latency of analytics like this to be a few hundred milliseconds. There’s no reason to need faster latency than that especially since the granularity of analytics is an hour. Microbatching gives high throughput and great fault-tolerance semantics while barely making the code more complicated.

The PState is a map from user ID to hour bucket to number of profile views in that bucket. There are about 8760 hours in a year, so it’s critical that inner map be subindexed. Subindexing enables inner data structures like this to be of arbitrary size, and you’re only limited by the amount of disk space on the machine.

Let’s now take a look at the topology code:

Like all the other topologies, this code is extremely straightforward. It’s straightforward because RamaSpace is straightforward, and Rama can express applications close to how you describe them by cutting out all the usual complexity.

While the write to the PState could use a localTransform, using a compoundAgg here is more concise. Aggregators automate routine things like initializing non-existent subvalues. In this case, the Agg.count() aggregator initializes the value for a bucket to 0 the first time it appears. You can read more about aggregators and other features they have here.

The last piece of the RamaSpace implementation is the resolvePosts query topology. Rendering a page of posts on a user’s wall requires not just the posts themselves but also information about the users who made those posts. While you could query all this user info client-side for every post, that would require over twenty round-trips with the module to get all the information needed to render one page of posts. It’s far more efficient to perform all that work in a single round-trip with a query topology.

With a query topology you specify a parallelized computation using the exact same API as used for making ETLs. This computation can query any or all of the module’s PStates and any or all of the module’s tasks. Here is the code for resolvePosts:

At a high-level, this code fetches up to the next 20 posts for "*forUserId" starting from "*startPostId". The first page would use a starting post ID of 0, which is the lowest possible ID. For each of these posts it partitions by the user ID to fetch their display name and profile pic, and then it combines all this information into a map to return back to the caller. To get the next page of posts, the client would invoke the query topology again with a starting post ID of one more than the highest post ID from the last page.

Query topologies are batched computations, a concept that hasn’t been introduced yet. So you’ll have to do a little more reading to fully understand this code. See this page for all the details on making query topologies.

Lastly, here’s how this query topology is used client-side. First, here’s how a handle to the query topology is created in the constructor of RamaSpaceClient:

Then, here’s how it’s invoked:

As you can see query topologies are invoked just like normal functions, except they perform highly scalable distributed computations to do their work.