Rollup And Pre-Aggregates

Rollup Example

Given the series above, lets store the sum and count with an interval of 1h.

Notice that all timestamps align to the top of the hour regardless of when the first data point in the interval "bucket" appears. Also notice that if a data point is not present for an interval, the count is lower.

In general, you should aim to compute and store the MAX, MIN, SUM and COUNT for each time series when storing rollups.

Averaging Rollup Example

When rollups are enabled and you request a downsampler with the avg function from OpenTSDB, the TSD will scan storage for SUM and COUNT values. Then while iterating over the data it will accurately compute the average.

The timestamps for count and sum values must match. However, if the expected count value for a sum is missing, the sum will be kicked out of the results. Take the following example set from above where we're now missing a count data point in ts2.

The resulting avg for a 2h downsampling query would look like this:

Pre-Aggregate Example

Given the example set at the top, we may want to look at the total interface traffic by colo (data center). In that case, we can aggregate by SUM and COUNT similarly to the rollups. The result would be four new time series with meta data like:

Notice that these time series have dropped the tags for host and interface. That's because, during aggregation, multiple, different values of the host and interface have been wrapped up into this new series so it no longer makes sense to have them as tags. Also note that we injected the new _aggregate tag in the stored data. Queries can now access this data by specifying an _aggregate value.

Now for the resulting data:

Since we're performing a group by aggregation (grouping by colo) we have a value for each timestamp from the original data set. We are not downsampling or performing a rollup in this situation.

Problems

Because of the (essentially) stateless nature of the TSDs, they likely won't have the full set of data available to perform pre-aggregates. E.g., our sample ts1 data may be written to TSD_A while ts2 is written to TSD_B. Neither can perform a proper group-by without reading the data out of storage. We also don't know at what time we should perform the pre-aggregation. We could wait for 1 minute and pre-aggregate the data but miss anything that came in after that minute. Or we could wait an hour and queries over the pre-aggregates won't have data for the last hour. And what happens if data comes in much later?

Additionally for rollups, depending on how users write data to TSDs, for ts1, we may receive the 12:15 data point on TSD_A but the 12:30 value arrives on TSD_B so neither has the data required for the full hour. Time windowing constraints also apply to rollups.

Solutions

Using rollups and pre-aggregates require some analysis and a choice between various trade-offs. Since some OpenTSDB users already have means in place for calculating this kind of data, we simply provide the API to store and query. However here are some tips on how to compute these on your own.

Batch Processing

One method that is commonly used by other time series databases is to read the data out of the database after some delay, calculate the pre-aggs and rollups, then write them. This is the easiest way of solving the problem and works well at small scales. However there are still a number of issues:

• As data grows, queries for generating the rollups grow as well to the point where the query load impacts write and user query performance. OpenTSDB runs into this same problem when data compactions are enabled under HBase.
• Also as data grows, more data means the batch processing time takes longer and must be sharded across multiple workers which can be a pain to coordinate and troubleshoot.
• Late or historical data may not be rolled up unless some means of tracking is in place to trigger a new batch on old data.

Some methods of improving batch processing include:

• Reading from replicated systems, e.g. if you setup HBase replication, you could have users query the master system and aggregations read from the replicated store.
• Read from alternate stores. One example is to mirror all data to another store such as HDFS and run batch jobs against that data.

Queueing on TSDs

Another option that some databases use is to queue all of the data in memory in the process and write the results after a configured time window has passed. But because TSDs are stateless and generally users put a load balancer in front of their TSDs, a single TSD may not get the full picture of the rollup or pre-agg to be calculated (as we mentioned above). For this method to work, upstream collectors would have to route all of the data required for a calculation to a specific TSD. It's not a difficult task but the problems faced include:

• Having enough RAM or disk space to spool the data locally on for each TSD.
• If a TSD process dies, you'll either loose the data for the aggregation or it must be bootstrapped from storage.
• Whenever the aggregation calculations are taking place, overall write throughput of the raw data can be affected.
• You still have the late/historical data issue.
• Since TSDB is JVM based, keeping all of that data in RAM and then running GC will hurt. A lot. (spooling to disk is better, but then you'll hit IO issues)

In general, queueing on a writer is a bad idea. Avoid the pain.

Stream Processing

A better way of dealing with rollups and pre-aggregates is to route the data into a stream processing system where it can be processed in near-real-time and written to the TSDs. It's similar to the Queuing on TSDs option but using one of the myriad stream processing frameworks (Storm, Flink, Spark, etc.) to handle message routing and in-memory storage. Then you simply write some code to compute the aggregates and spit the data out after a window has passed.

This is the solution used by many next-generation monitoring solutions such as that at Yahoo!. Yahoo is working to open source their stream processing system for others who need monitoring at massive scales and it plugs neatly into TSDB.

While stream processing is better you still have problems to deal with such as:

• Enough resources for the stream workers to do their job.
• A dead stream worker requires bootstrapping from storage.
• Late/historical data must be handled.

Share

If you have working code for calculating aggregations, please share with the OpenTSDB group. If your solution is open-source we may be able to incorporate it in the OpenTSDB ecosystem.