Hundreds of terabytes in Athena: partitions by the hour, and queries that do less

At an Italian fintech in cybersecurity, an internal alerting service had one job: tell a subscriber when their data shows up in a new leak. The data it searched was hundreds of terabytes of Parquet on S3, queried with Amazon Athena, and Athena bills by the data each query scans. Serverless or not, the big queries never finished: they ran into Athena’s time limit and were cancelled. A query runs for 30 minutes by default, and 240 minutes is the most the limit can be raised to (AWS: Athena service quotas); ours were hitting 240. After the changes below the same queries ran in about 30 minutes.

Two directions of the same question

Alerts came from two kinds of jobs:

• New files, existing subscribers. A fresh batch of leaked data arrives; every current subscriber has to be checked against it.
• New subscribers, existing files. Someone signs up; everything already collected has to be checked against their data.

The first job reads a small, recent slice of the data. The second reads all of it, for a handful of people. The layout on S3 has to serve both, and as it turned out, each job was fixed by a different change.

CSV files with millions of records each were ingested by a service written in Rust, processed by analysts, and landed on S3 as Parquet. The alerting jobs queried that data, checked whether a record belonged to a subscriber, and created an alert when it did. Click a box for details, or switch to Before to see where it broke.

Partitions down to the hour

Athena reads only the partitions a query filters on. AWS’s own example shows a date filter cutting one query from 74.1 GB scanned to 29.96 MB (AWS Big Data Blog).

The data was partitioned by date; I added the hour. A job checking a new batch then reads the hours that batch arrived in, not the whole day. That helps the first job only: the second one reads every partition anyway. New partitions were registered with MSCK REPAIR TABLE, which finds new dt=…/hour=… prefixes on S3 and adds them to the catalog.

CREATE EXTERNAL TABLE records (  -- [columns])1PARTITIONED BY (dt string, hour string)STORED AS PARQUETLOCATION 's3://[bucket]/records/'; -- after new files land on S32MSCK REPAIR TABLE records; -- a new-files job reads one hour, not the whole tableSELECT ...FROM records3WHERE dt = '[yyyy-MM-dd]' AND hour = '[HH]';

Queries that do less

Partitions do nothing for a job that reads everything. For that one, what mattered was the shape of the data and the SQL that reads it.

One row per value, and the file’s metadata on every row

Each record held a single value, such as a first name, together with the metadata of the file it came from. A CSV with millions of rows meant the same metadata copied millions of times, and there could be hundreds of metadata fields: spread over many columns, or packed into one.

Packed is worse than it sounds. Sometimes the column was a struct, which Athena (built on Presto, now Trino) reads directly. Sometimes it was the same object stored as a string, and then every query had to parse that string, row by row, before it could filter on a single field.

So the metadata moved to a table of its own: one row per file, and in the records a single file_info_id in place of hundreds of repeated values. The same query before and after, with a Diff between them:

SELECT valueFROM recordsWHERE dt = '[yyyy-MM-dd]'1  AND json_extract_scalar(file_info, '$.source') = '[source]'; -- parsed on every row

SELECT r.valueFROM records r1JOIN file_info f ON f.file_info_id = r.file_info_idWHERE r.dt = '[yyyy-MM-dd]'2  AND f.source = '[source]'; -- a typed column, read once per file

Two things changed. The filter on file metadata now runs on a small table with one row per file, not on billions of rows, and the JSON parsing that ran on every row is gone. Then the few matching files are joined to the big table, with the small table on the right of the join, as AWS recommends (AWS Big Data Blog).

Named steps

Long nested queries became a sequence of CTEs, one named step each, so it was clear which step did the expensive work.

When a query fails, ask for less

Even with the layout fixed, a query over a long range could still hit the time limit. Instead of retrying the same query, a failed range was split into smaller ones and the parts ran in parallel: a month into four weeks, a week into days, a day into hours. The total work stays the same, but each piece fits under the limit, and running them side by side brings the wall-clock time down too. The simulation below runs that logic on a month of data (illustrative, not measured):

type Range = { from: Date; to: Date; unit: 'month' | 'week' | 'day' | 'hour' }; const finer = { month: 'week', week: 'day', day: 'hour' } as const; async function runRange(range: Range): Promise<void> {  try {1    await runAthenaQuery(range); // one query for the whole range  } catch (error) {2    if (range.unit === 'hour') throw error; // nothing smaller left3    const parts = split(range, finer[range.unit]); // a month becomes weeks, a week days, a day hours    await Promise.all(parts.map(runRange)); // the parts run in parallel  }}

Only the part that fails gets split. A month that runs in one query stays one query; a heavy day inside it is the only one that goes down to hours.

What it changed

If an Athena job is slow or failing, look in this order: does the query filter on the partitions, does it parse or compute something on every row that could be done once, and does it filter the small tables before joining the big one. When a range is still too big, split it on failure rather than retrying it whole.