Scale By The Bay 2021 : Tudor Bosman, Karen Li, How We Built SQL Rollups on Streaming Data
thank you uh thanks thanks thanks all of you for joining us uh i'm tudor um i'm going to be going fairly quickly to mainly the first part of the presentation um i'm going to introduce myself i'm going to introduce the rockset and then i'll hand it over to karen and i'll stick around for the questions and answer section at the end so um as i said i'm tudor we are rockset um yes uh workset is a real-time analytics database in the cloud we we connect to data streams we are able to ingest data from a variety of sources some of those are data streams like kafka or kinesis some are ltp databases some are data lakes like s3 or gcp and yeah we are ingesting the data and then we provide full feature sql support on it for analytics we uh the data does not need to conform to a very strict schema we are pretty much schema-less so if you know for example if a document if some field is missing from some documents or has different types among different documents that is something that we can do we index everything three times and that's going to be on the very next slide um we in essentially we have a column store which makes us similar to warehouses in that we have optimized for lives for long scans for live scans uh we have a go store which i'll talk about in a second and we have a search index which allows us to be efficient for uh high selectivity queries like or where product id equals something and of course the row store is used as in traditional databases to retrieve entire documents that have met the search query and i believe that's basically all i would have to say i will be around for the q a uh kevin is going to take it over from here karen did most of the implementation for roll ups so she is absolutely the best person to speak um awesome thank you so much tutor for the intro so i'm karen and i'm going to be talking a bit more about rollups um so we're going to start off with why why did we build roblox in the first place so rollups are a way for you to pre-aggregate and transform your data at ingest time and so how it works is that when you create a collection in rockset if you want to use rollups what you need to do is specify one single sql query which is a rollup query and you can think of the rollup query as a standing query that runs continuously on your incoming data and transforms it um you know as the data streams in so most likely your roll of sql query is going to look something like what we have here so you're going to be grouping by a couple of dimensions and then running some aggregation functions on on your other fields and then the from input part of this query is standard to all real queries it's basically just representing the input data stream so the reason why we built rollups is to make it more efficient to form analytics on streaming data so you can imagine that you know there could be times where you have a lot of data streaming in and so it's really expensive to index and store the entire raw data set and you know in other cases maybe you already know that you only want to perform analytics on an aggregated set of your data and so why not do that at ingest time and so if you do do that in jaztime with rollups the benefits you get are that you will reduce the storage size and then you also get improved query performance because you've essentially pre-computed the fields that you care about so let's look at an example of how you could use roll-ups so in this case let our data source be a net network of cameras that detects cats and dogs so you know periodically each of these cameras is going to emit an event of the following form your event is going to contain a time stamp a location and the number of cats and dogs seen so this data stream is streaming in and you want to you know perform some analytics on this so let's use the following role of query as an example looking at it from top down uh we can see that first what we're doing here is we're massaging our timestamp field a little bit and putting in an output field called event time we also select location because we're going to be grouping by event time and location and then we compute a bunch of metrics so we take the sum of the cat scene we do the animal count high water mark using the max aggregation function and then at the very end we have my crazy metric which is a pretty contrived educational example of how complex your roll-up queries can get so here we're combining the results of three aggregate functions count star average and approx distinct uh using some mathematical functions so here we can see what our roll of query is actually going to do to our raw data stream so in the source data preview section that's the raw data stream and then the transformed preview uh it's going to show you what happens to your data after it's you know passes through the roll of query so here we can see we just have one input row of data that's going to you know map to one aggregation group and so we end up with one output row things get a little bit more interesting when our second row of data appears and in this case the second row of data maps to the same aggregation group as the first one and so what the role of query is going to do is actually merge and aggregate these together and of course your metrics get updated appropriately now the row of data comes in this time it's going to a different aggregation group because the location in los angeles is different and so now we have two output rows in our roll-up collection another row uh belonging to the second aggregation group and so we still have two upper rows of data but our aggregation functions their values have been updated and then finally we get one more row of data and this one belongs to a third aggregation group because uh the time and our granularity is different and so at the very end our roll-up collection is going to contain three rows of data so now let's look at some of the design principles we used when creating roll ups so there are three main design decisions and looking at them one by one uh the first one is that we wanted to keep a sql interface for rollups uh to make it easier for users to start using rollups so you don't need to learn any other language to define your rollups you can just continue using sql the second point is that we wanted to keep the real-time nature of your data even when you use roll-ups and so the traditional ahead of time aggregation approach means that you actually don't have access to your data until the end of the application interval because you need to look at all the data to know how to merge it correctly rockstar takes a different approach where we actually do allow updates to aggregation groups and this allows you to you know query your role of collection in real time and see the results as the aggregations are happening so this brings us to our third point of handling out of order arrivals and because we said that we do allow updates to aggregation groups uh handling out of order arrivals just works so this information on this slide is a bit redundant but basically we're just going to reiterate that with the real-time aggregation that rocks it provides you're able to update your aggregation groups which allows you to have your data be available in real time and we can trivially handle out-of-order arrivals so we all know that there is no such thing as a free lunch and so the trade-off of real-time aggregations is that it requires a more complex implementation under the hood one part of this complexity is that we must store the partial states of aggregations so the partial state of an aggregation is the state you need to maintain to compute the final value of your aggregation function and it also has a property that it can be updated incrementally so here we have two aggregation functions as our examples average and max so looking at average first we can see that the partial state here is actually two values the sum and the count and then the final value or final state is going to be the average and then with max the partial state is the max value seen so far and the final state is the exact same so an interesting point to note here is that the partial state of an aggregate function may or may not be the same as its final state another change of the design with rollups is actually how we index these real-life collections so like tutor was saying before we have converged indexing which means that we store all the data in three different indexes so the row store and column source stay the same and the only change is in the search index so with rollups in the search index we index all fields except for the ones where the aggregate function value has a different partial state from the final state and the reason for this is that when you do a look up in the search index you're going to be using the final value of the aggregate function and if the partial state that's stored is not the same as the final state you won't really be able to perform that lookup and so there's no point in storing it in the index next let's look at the implementation and some of the challenges we face there so first i'm going to be talking about how we decompose the rollup sql query i've mentioned that to use rollups you specify one single sql query which is the roll up query and so it's interesting that conceptually this is actually two separate queries the ingest query which contains a logic that happens at ingest time and then the row query which contains a logic that happens like query time at ingest time what we do is compute the partial state of aggregate functions and we also compute simple expressions which in our context just means expressions that don't involve aggregate functions in addition we'll also execute the where clause to filter out unwanted input data rows and of course we run the group by clause to merge aggregate groups so that you know multiple input rows that go to the same aggregation group will just form one output row then at query time we need to finalize aggregate functions where finalize is the process of converting the partial state of an aggregate function to the final state so you know for average this is the time where we actually compute the division of the sum in account we also need to compute complex expressions which are expressions that do involve aggregate functions and finally the having clause also needs to be executed at query time because it requires performing a filter on an aggregate function value so going back to our example from before let's look at how we can decompose this into the ingest query and the wrap query so everything that's highlighted will be executed in jest time as part of the ingest query so we can see that we're able to compute our event time field because it doesn't rely on any aggregate function values similarly the location field just passes through directly from the input and then we have to compute the partial states of our aggregate functions and at the very end of course the group i is going to merge input documents that map to the same aggregation group to make our example a little more concrete uh here on the left we have some raw input documents and then on the right we can see how they get transformed by the ingest query and uh these documents are going to contain a band time in the location fields which should be executed and then it's also going to have the partial states of our aggregate functions next uh the wrap query shows us what's going to run at query time and for us we only need to finalize two aggregate functions average and approximate and then we're going to you know finally compute my crazy metric and so on the left we have the documents we were just looking at that were transformed by the adjust query and then they'll pass through and get transformed by the wrap query and so the documents transformed by the wrap query are the final user visible documents that you see when you query your role of collection so event time and location which we already had before stay the same and then we'll see that we actually uh you know compute the final values of all of our metrics which are cat count animal account high water mark and my increasing metric next we'll talk about how we execute sql at an ingest time or in other words how do we actually run the ingest query part of the roll of query so first taking a step back let's look at the right path and rock set without roll-ups so customer data is going to be living in a variety of data sources and it's going to make its way into rockset through the taylor service taylor service contains the logic needed to extract data from these different data sources and it's going to write the raw input documents into a distributed log which serves as our write ahead lot then a different service called the leaf is going to tail documents from the distributed log and index and store them in roxdb so the leaf service maintains the rocksbb instance that's needed for both serving rights as well as reads at query time with rollups uh the picture gets a little bit more complicated and the difference is that now we've inserted a new service called the aggregator which sits in between the taylor and the distributed log so of course the tailor has the raw input documents and it's actually now going to pass them to the aggregator the aggregator will run the ingest query on the data and return the transform documents to the tailor and then the taylor will write those transform documents to the distributed log where the leads will pick that up and index them in roxdb so let's zoom in in a bit to uh the rpc between the taylor and aviator services so on the left we have the raw documents that get sent from the tailor to the aggregator uh we have a new rollover example just for the sake of space so everything fits on one slide and here we're just going to be grouping by location and then computing uh the partial state of the average function and so you can see in the transform documents that they've been grouped by location and we have the partial state of the average and then these transform documents get sent back from the aggregator to the tailor so here's our same diagram again now it's been annotated with a few diagrams um so the diagrams on the aggregator service show that the raw input data could get merged um when we run the ingest query and you know multiple rows of input data mapped to the same aggregation group and then you'll also see that on the leaf there's a diagram and what this is saying is you know it's kind of a spoiler for the next section but we actually do need to do some merging on leaves as well in some cases and the reason why we have to do this is that um you know to keep the data real time the leaf is going to be able to serve reads for its existing aggregated roll-up roll-up state but if some document comes later out of order we still need to be able to update the state on the leaf appropriately so yeah we're going to talk about how we handle these out of order arrivals and like we were just saying the lease service needs to be able to update existing aggregation groups so let's say the state on our leaf is just these two documents we're grouping by location and then we tail a new document from the distributed log on the leaf and this document belongs to the same aggregation group as one of the existing ones because location san francisco is the same and so the leaf needs to be able to update that aggregation group so the way we handle updates to an existing aggregation group on the leaf takes advantage of an existing principle in rockset which is the underscore id special field the underscore id special field is the primary key in a roxa collection and every single document in every roxa collection has the underscore id field and because it's the primary key we give you ups or right semantics based on the value of that field and so if you ever write that's you know going to the same document because it has the same underscore id value then we'll actually update your existing document for rollups what we do is we set the value of the underscore id field to the 128-bit hash the grouping dimensions and by doing this we establish the relationship that one roll-up document represents exactly one aggregation group so going back to our favorite example here we see that we're grouping by event time and location and so what this means is that every document in this role of collection is going to have underscore id field with the value of the 128-bit hash of the event time and location and so when you get updates to the same aggregation group you'll know that they're actually going to the same roll-up document because the underscore id value is the same and so we take advantage of the existing observed semantics and the way we actually update your existing role of document is just using a read modify write approach and i'm going to note here that the only values we need to update in your roll-up document are the aggregation function partial states and that's because the only other fields that could exist in your old document are the grouping dimensions and by definition those will not change so what we do is we load the aggregation function partial state into memory from roxtv we update the partial state in memory and then finally we write that back to roxdb so the logic that's used for merging aggregation partial states here is actually the same logic we use in distributed aggregations at query time so going back to our average and max examples um for average what we do is we take the sum of the sum and count fields independently to merge the partial states and then for max we just take the max the two input values to you know get the final uh merge state and so the logic for merging partial states for different application functions will different needs to be defined uh independently next let's talk about right semantics so by default rockset supports at least once right semantics and if all of your updates are item potent then that's totally fine it doesn't matter if a right gets applied multiple times you know that your final state is always going to be the same on the other hand rollups need exactly one's right semantics because if a right gets applied multiple times you could end up with double counting issues and you know your aggregate function values will be incorrect our implementation of exactly once right semantics uh comes from our mental model of how streaming data sources work and so our mental model is that we have rights arriving in n independent streams and within a given stream the rights are ordered so far exactly once right semantics implementation we assign a unique update id identifier to every single write the update id is composed of two separate fields the first is the shard id so which shard does this right belong to and then we have the sequence number because we know that um you know the rights are ordered within the stream and this sequence number is monotonically increasing so in practice the value of the short id is going to be your partition or kinesis chart and then the sequence number is going to be the offset so now that every single right has an update id the leave service will enforce exactly once right semantics by rejecting rights for a given charge that have a sequence number that is less than or equal to the highest sequence number we've seen so far for that chart um and because you know if we've seen if we see that the sequence number is lowered and the highest that we've seen so far we know that that right has already been applied and so we we don't want to apply it again the metadata for the mapping from sharp to high sequence number seen is persistent in roxdb and we can prune this metadata on a per chart basis whenever we know that a shard is never going to get any more updates so that was a lot of information let's just summarize the implementation again so looking at the right path first the tailors will send the raw input documents to the aggregators where the aggregated services run the ingest query and return transformed documents then the tailor will write these transform documents to the distributed log so the leaves can tail and index them and then of course the leaf needs to be able to update uh roll up documents as needed and throughout this whole process we need to make sure that rights are processed exactly once to avoid any double counting issues then on the read path you can think of querying a roll-up collection like querying a view where the wrap query is the view query and what the raw query does is finishes the computation of the roll query using the values pre-computed in just time so we're going to finalize all of our aggregate functions uh you know for average we'll compute the sum divided by the count and you know we'll compute all of our complex metrics and then just just an interesting implementation detail is that you know we've said that conceptually acquiring a role of collections like creating a view um but actually the implementation for uh the views and current world collection is the same and if you want to do it yourself um all you need to do is create an account at roxas.com and try out sql roll-ups on your streaming data cool so that's all the content we have um and i think now it's time for the q a should we switch over to special chat or or just stay here it does not appear like there are any questions in the discord right now so i want to thank you some really cool stuff you guys are working on and then we can have everybody switch over to this facial chat and q a there okay sounds good um i guess we'll see you all there thank you