Papers
Gorilla: A Fast, Scalable, In-Memory Time Series Database
Tuomas Pelkonen, Scott Franklin, Justin Teller, Paul Cavallaro, Qi Huang, Justin Meza, Kaushik Veeraraghavan · VLDB (Proceedings of the VLDB Endowment, Vol. 8, No. 12), 2015
The Problem: Monitoring Millions of Metrics per Second
Facebook needed to monitor the health of thousands of machines spread across several data centers, and for that it used a time series database (TSDB) built on HBase, part of the system called ODS. In early 2013 the team realized that this storage would not withstand the future read load: the 90th percentile of queries already took several seconds, which stalled alert automation, and users avoided interactive analysis because larger queries took tens of seconds or timed out. The usage profile was peculiar. Writes dominate: the rate could exceed tens of millions of points per second, while reads are orders of magnitude less frequent, coming mostly from dashboards and automated alarm systems. In addition, the authors observed that at least 85% of ODS queries asked for data from the last 26 hours. The design insight is that people monitoring systems hardly care about an individual point, but about aggregations, and that recent data is worth much more than old data, because knowing something is broken right now is more useful than knowing it broke an hour ago. Hence the proposal: use an in-memory database as a write-through cache for the most recent data, leaving HBase for long-term storage. The listed requirements include 2 billion series, 700 million points per minute, and reads in under a millisecond.
Compressing Timestamps with Delta-of-Delta
To fit billions of series in memory, Gorilla compresses each series separately, but as a stream, point by point, using only information from the previous points. Each original point is 16 bytes (a 64-bit timestamp and a 64-bit double value). As for timestamps, the authors noticed that the vast majority of data arrived at a fixed interval, for example one point every 60 seconds, sometimes one or two seconds late or early. Instead of storing the full timestamp, Gorilla computes the difference between consecutive timestamps (the delta) and then the difference between successive deltas (the delta-of-delta). If the deltas are 60, 60, 59 and 61, the deltas-of-delta are 0, -1 and 2. The encoding is variable-length: a single '0' bit when the delta-of-delta is zero; header '10' plus 7 bits for values between -63 and 64; '110' plus 9 bits for the range -255 to 256; '1110' plus 12 bits for -2047 to 2048; and '1111' plus 32 bits for the rest. The block header stores the starting timestamp, aligned to a two-hour window, and the first delta takes 14 bits. These limits were chosen by sampling real series. In a sample of 440 thousand real timestamps, about 96% fit in a single bit. When a point is missing, the delta-of-delta still falls in the smaller ranges, so losing one data point does not destroy the compression.
Compressing Values with XOR and the Result: 1.37 Bytes per Point
For floating-point values, Gorilla drew inspiration from compression schemes used in scientific computing, but swapped the expensive prediction for something simple. The authors saw that in many series the value changes little between neighboring points, and that many sources record only integers. When two doubles are close, the sign, exponent, and first bits of the mantissa match, so the XOR between the current value and the previous one has many leading and trailing zeros, with the relevant bits clustered in the middle. The encoding is: the first value is stored uncompressed; if the XOR is zero (the value equals the previous one), a single '0' bit is stored; otherwise, '1' is stored followed by a control bit. With control '0', the significant bits fit within the same block of zeros as the previous value and only they are stored. With control '1', the number of leading zeros (5 bits), the length of the significant segment (6 bits), and the significant bits are stored. In a sample of 1.6 million real values, more than half fit in one bit, about 30% use the control case '10' (average of 26.6 bits), and about 19% use '11' (average of 36.9 bits). Together, the two techniques brought the average cost from 16 bytes to 1.37 bytes per point, a 12x reduction. There is a trade-off: larger blocks compress better, but force short queries to decode more data, and beyond two hours the gain is small.
Memory as a Write-Through Cache and Persistence without ACID
Gorilla works as a write-through cache of the last 26 hours of data, written durably to HBase for the long term. The data lives in memory in a structure called TSmap, made up of a vector of pointers to series and a case-insensitive map from series name to the same series. The vector allows efficient paginated scans, and the map allows constant-time lookup. Each series is a sequence of closed blocks (data older than two hours, which never changes again) and a single open, append-only block that receives the newest points. The series key is a string name, and sharding by that key allows scaling just by adding machines: at launch, the 26 hours of data fit in 1.3 TB of RAM distributed across 20 machines, and later the cluster was doubled twice, reaching 80 machines. To survive the loss of a host, data is written to GlusterFS, a POSIX-compatible distributed file system with 3x replication. Each shard keeps an append-only log, buffered up to 64 kB (normally one or two seconds of data), and every two hours it copies the compressed blocks to a block file. The authors state that Gorilla does not offer ACID guarantees and that this log is not a traditional write-ahead log: they accept losing a little data in a crash in exchange for higher throughput and write availability.
Availability over Consistency: Regions, Failures and Tools
The fault tolerance design starts from an explicit priority: when there is a conflict, favor the availability of recent data over the completeness of old data, since HBase serves historical queries and automatic detection systems work well with partial data, as long as it is recent. To survive the loss of an entire region, Gorilla keeps two fully independent instances in distinct regions. Each write is sent to both, with no attempt to ensure consistency between them, and reads go to the nearest available instance. If a region goes down, reads are attempted on the other. Within each region, a Paxos-based system called ShardManager assigns shards to nodes. During shard movement, write clients buffer up to 1 minute of data and discard the oldest points. After the failure, a shard is partially unavailable until it is reread from disk, and the query returns a result marked as partial. The client library then tries the other region and, if both return a partial result, hands it to the caller with an incomplete-data flag. The authors also report new tools made possible by the low latency, such as a time series correlation engine that does brute-force search over up to 1 million series at a time.
See the ideas at work
Practice the designs that use this paper in the System Design school.