S3 is slow for small reads. We built a telemetry data lake on it anyway

Traditionally, it has been quite straightforward - if you want a fast system, ensure your software is designed to keep the primary data in memory or on a local disk. Users would just need to run the software on machines sized to hold everything in memory / disk.

There are headwinds that are challenging this basic assumption. First of all the data volumes are growing across the board. At the same time, the compute and memory prices have grown several times in the last year.

The traditional in-memory / on-disk doesn’t work anymore. So, we started building Parseable, ~4 years ago with object store as the primary storage target. This means Parseable is a stateless layer on top of an object store and offers a complete set of observability featureset.

We come across the question frequently - “Isn’t Parseable slow because of object store?”. Physically, object stores are one of the slowest systems, especially when reading small files.

Memory for example takes ~100 nanoseconds. A locally attached NVMe drive takes ~100 microseconds. But if you ask S3 for a small object, you're looking at tens of milliseconds to get the first byte, and you always pay for each request.

In this post we analyse - how slow really is object store? What is the latency? How does it vary based on query patterns? And finally, can we intelligently manage cache to ensure a reliable, scalable and fast observability platform?

We start with a production grade Parseable cluster, run the same three queries against it with the data coming first from the bucket and then from local SSD, and measure where the time goes.

We set up a Parseable cluster on Google Cloud Storage, querying logs-dataset. We have timed the same queries with the data being read from the bucket and from the local SSD.

Everything in this post is from this cluster. It is a distributed Parseable cluster, comprised of 3 nodes, one of each type:

Let’s take a common query. It groups errors by the message in logs-dataset. We’ll use this same query later in the post as well:

SELECT "body_message", count(*) AS errors FROM "logs-dataset" WHERE severity_text = 'ERROR' GROUP BY "body_message" ORDER BY errors DESC

When you run this on Parseable, on a cold query node (with nothing cached locally), this query would issue three classes of requests to object storage, in this order:

Parquet footer acts as a miniature index for the file it belongs to, since each column chunk's min and max can be used to eliminate an entire file or row group before reading any of its data. All those files not inspected are a set of requests that will never be made.

Each ingestor writes roughly one parquet file per minute of data, so the number of files a cold query has to consider grows with the length of its window:

The values are calculated at 24-hr intervals since the test. 1hr and 7-day figures are based on estimates. GET Range requests dominate due to the large number of row groups per Parquet file and the fact that the query is reading byte ranges of the relevant column chunks. The low number of GET requests is because some objects are read in their entirety (such as manifests). Both GET and GET Range requests are priced identically by S3.

There are three figures of interest in these numbers: requests × latency, divided by concurrent requests, is wall-clock time; requests × price is the bill; bytes fetched is the bandwidth. The bill is the least of our worries: at $0.0004 per 1,000 GET requests to S3 Standard in us-east-1, the roughly 1.125 million GET and ranged GET requests in the 24-hour test would cost about 45 cents. The latency is the challenge.

This is where parallelism comes in. The query plan splits the scan into partitions, four per CPU core, so 64 on our 16-vCPU querier, and makes requests 64 at a time instead of one after another. The 24-hour cold query made about 1.12 million ranged GETs and 25 whole-object GETs. At ten milliseconds per request, issuing them one by one would take about 3.1 hours. Across 64 parallel streams, that becomes roughly 17,577 rounds, or 176 seconds. This is a back-of-the-envelope upper bound: range merging and overlapping network reads with query execution reduce the observed wall-clock time. Even so, object-store latency remains part of the cold-query path, which is exactly what the Smart Cache removes from most queries.

Parallelism has a ceiling. S3 allows at least 5,500 GET requests per second per prefix, and a GCS bucket starts at about 5,000 reads per second. Past that, requests get rejected and retried.

A query node can contain two different kinds of local copies: in-memory (Parquet metadata cache) and on-disk (Smart Cache).

Manifests are an exception in that they are read from the object storage for each query, independent of the cache. They are also small, and there is one per ingestor (per day in the query window), so the count is the number of ingestors times the number of days (as opposed to files).

This represents a difference from other systems that keep the primary data of the system in memory or local disk. Local storage is a copy (and must keep everything that may be queried) and also has to be copied elsewhere (because losing the disk would mean losing the data). To Parseable, local storage is a cache, which you can size for speed, not safety, and which you can throw away at any time.

Even if cache has recent data, it is only useful if most of the queries are in the recent time range - essentially how do we optimize for cache hits. Let's look common queries:

They all look at the last few hours. The old data is typically queried for a postmortem or a report that happens once a month, but not as often.

We are not the first to notice this. When Facebook was building Gorilla, the in-memory time series database that powered its monitoring, they discovered that at least 85% of all their monitoring queries were for data gathered in the previous 26 hours. So they kept the last 26 hours in-memory in Gorilla, and it's the same reason a cache that spans the last day is useful to answer most queries.

Queries do not have to be all-local or all-remote. When a query goes past the cache, as a seven-day query does with a 24-hour hot tier, recent files are read from local SSD and older files are fetched from the bucket, in the same query. The cache shortens the part of the query that's most likely to be repeated. One important note here is that newly ingested data is not in the hot tier until it has been committed to object storage and synced to the query node.

We have selected three queries that ask progressively more of the query node, ranging from virtually none to a filter and a group-by on a text column. Each one spans the last 24 hours of logs-dataset, a time interval that would be typical for a dashboard or an on-call engineer's needs:

-- 1. Count all rows SELECT count(*) FROM "logs-dataset" OFFSET 0 LIMIT 500

-- 2. Count rows where the timestamp is set SELECT count(*) FROM "logs-dataset" WHERE p_timestamp IS NOT NULL

-- 3. Errors grouped by message SELECT "body_message", count(*) AS errors FROM "logs-dataset" WHERE severity_text = 'ERROR' GROUP BY "body_message" ORDER BY errors DESC

The first one does not do any work other than the data coming in, so it's just a measure of how fast data reaches the engine. The second one has to read the p_timestamp column and validate it for every row. The third one reads severity_text and body_message, filters on one, and groups on the other, and in addition, the body_message is free text, so it's the most expensive of the three.

We ran each query in three cache states on the querier:

Each state takes one class of request off the network. Here's where every request is served from:

Cold is the S3 path from the beginning of this post. Hot is the path most real queries take. In our hot runs, the hot tier held the full 24 hours the queries asked for, so almost every column chunk was served from local SSD, and only the manifests went to the bucket. Warm is in between, and teaches us whether caching just the footers is sufficient.

Every query ran three times in each state, and we report the median.

Here’s the same table as a picture. Each bar is a query’s median time as a share of its cold time, and the thin line shows the fastest and slowest run:

The key to success was copying the data over from S3 to the local SSD. With the hot tier enabled, queries that count all rows in the last 24 hours improved from 14.0 seconds to 3.8, which is a 73% improvement. The heaviest query, errors grouped by message, improved from 48.4 seconds to 35.9. Caching the footers alone didn't move any of the three queries outside the run-to-run noise.

Reading from SSD takes the wait out. Counting all the rows is nearly all the cost of getting that data to the engine, and the hot tier eliminates that. It also made runs consistent: cold runs of this query varied from 9.1 to 15.6 seconds, while hot runs varied from 3.5 to 4.0. Reading from the bucket introduces its variance into every query. Reading from local SSD doesn't.

Heavier queries keep their compute. The group-by saved us 12.5 seconds, but there was still almost 36 seconds left. This is the cost of decoding, filtering and grouping on a free-text column, which no cache would alleviate. The hot tier contained the entire 24 hours which these queries spanned, so there was no time spent waiting on the bucket for data; only the manifests were delivered from there. What is left is the work of the engine, which would be the same on any storage.

The metadata cache alone was lost in the noise. In the warm state, medians moved between 2% slower and 25% faster. The 25% is the simple count, and even there the warm runs (9.1 to 12.9 seconds) overlap the cold ones (9.1 to 15.6). And across the same query, cold runs varied by up to 6.5 seconds from one run to the next. A footer is one tiny read per file, and besides the column-chunk reads and the work done by the engine after, taking them away didn't make a difference at this scale. The footers aren't where the time is. The data is.

The timestamp count fell between. It needs to read the p_timestamp column across all rows, so it performs more operations than the simple count, but far less than the group-by. From SSD it took 4.3 to 5.7 seconds, versus 9.4 to 13.5 from the bucket, 59% faster.

Combined with where observability queries fall, this answers the question in the title. S3 is as slow as ever, but a query against the last day (which is most of them) reads from the SSD.

The next question that comes to mind is: how much SSD does each dataset need?

The answer isn't “enough for your retention”. It's “enough for the window most of your queries hit”. Our logs-dataset takes in about 12 million events a minute. Over the last 24 hours, that's 17.3 billion events and about 900 GB of Parquet, roughly 650 MB a minute. So a hot tier sized to the last 24 hours needs around 900 GB of SSD for this dataset.

These figures grow with your ingest volume, so focus on the rule of thumb rather than the exact numbers: check the time ranges your dashboards and alerts use, and size your hot tier accordingly. Anything outside that window remains in object storage and stays fully queryable; it just costs a round trip.

In our tests, that round trip cost about 10 seconds on the simple count (3.8 s hot against 14.0 s cold) and 12.5 seconds on the group-by.

The metadata cache is sized separately, in memory, with DataFusion's P_PARQUET_METADATA_CACHE_SIZE setting. Remember that footers are a small fraction of each file, so caching the footers for your common query windows will necessarily take up far less memory than the Smart Cache will take up on disk. The easiest way to size this is to actually measure it: configure a generous cache (say, 5 GB), run a query over the window that your team uses most often, and when