Zapytania Athena przy limicie 240 minut: partycje co godzinę i zapytania, które robią mniej

Część zaplanowanych zapytań, przy których pracowałem, działała na dużym zbiorze danych w Parquet na S3, przez Amazon Athena, która nalicza opłaty za dane przeskanowane przez każde zapytanie. Serverless czy nie, te największe nigdy się nie kończyły: dochodziły do limitu czasu Atheny i były anulowane. Zapytanie domyślnie może trwać 30 minut, a 240 minut to najwięcej, do ilu da się ten limit podnieść (AWS: Athena service quotas). Te dochodziły do 240. Skończyć się pozwoliły im trzy zmiany: partycje aż do godziny, zapytania, które robią mniej na każdym wierszu, i dzielenie zakresu, gdy zapytanie się nie powiedzie. Ten wpis jest o technikach; tabele, kolumny i wartości w przykładach są poglądowe.

Dwa rodzaje zaplanowanych zapytań

Większość zaplanowanych zadań na rosnącym zbiorze danych zadaje jedno z dwóch pytań:

• Co nowego? Przychodzi świeża partia danych i trzeba ją sprawdzić względem wszystkiego, co system już wie.
• A co ze wszystkim? Coś nowego po drugiej stronie, nowa reguła albo nowy wpis, trzeba sprawdzić względem wszystkich zebranych dotąd danych.

Pierwsze czyta mały, świeży wycinek danych. Drugie czyta wszystko. Jeden układ na S3 musi obsłużyć oba, a każde naprawia inna zmiana. Kliknij pole, żeby zobaczyć szczegóły, albo przełącz na Przed, żeby zobaczyć, gdzie układ się psuje.

Partycje aż do godziny

Athena czyta tylko te partycje, po których zapytanie filtruje. Własny przykład AWS pokazuje, jak filtr po dacie zmniejsza skan jednego zapytania z 74,1 GB do 29,96 MB (AWS Big Data Blog).

Dane były podzielone na partycje po dacie; dodałem godzinę. Zadanie sprawdzające nową partię czyta wtedy godziny, w których ta partia przyszła, a nie cały dzień. To pomaga tylko pierwszemu zadaniu: drugie i tak czyta każdą partycję. Nowe partycje były rejestrowane przez MSCK REPAIR TABLE, które znajduje nowe prefiksy dt=…/hour=… na S3 i dodaje je do katalogu.

CREATE EXTERNAL TABLE records (  -- [kolumny])1PARTITIONED BY (dt string, hour string)STORED AS PARQUETLOCATION 's3://[bucket]/records/'; -- gdy nowe pliki trafią na S32MSCK REPAIR TABLE records; -- zadanie dla nowych plików czyta jedną godzinę, nie całą tabelęSELECT ...FROM records3WHERE dt = '[yyyy-MM-dd]' AND hour = '[HH]';

Zapytania, które robią mniej

Partycje nic nie dają zadaniu, które czyta wszystko. Dla niego liczył się kształt danych i SQL, który je czyta.

Jeden wiersz na wartość i metadane pliku na każdym wierszu

Każdy rekord zawierał jedną wartość razem z metadanymi pliku, z którego pochodził. Plik z wieloma wierszami oznaczał te same metadane skopiowane na każdy z nich, a pól metadanych mogło być wiele: rozłożonych na wiele kolumn albo spakowanych w jedną.

Spakowane jest gorsze, niż brzmi. Czasem kolumna była strukturą (struct), którą Athena (zbudowana na Presto, dziś Trino) czyta bezpośrednio. Czasem był to ten sam obiekt zapisany jako tekst i wtedy każde zapytanie musiało parsować ten tekst, wiersz po wierszu, zanim mogło przefiltrować po jednym polu.

Dlatego metadane przeniosły się do osobnej tabeli: jeden wiersz na plik, a w rekordach jedno file_info_id zamiast setek powtarzanych wartości. To samo zapytanie przed i po, z widokiem Różnice między nimi:

SELECT valueFROM recordsWHERE dt = '[yyyy-MM-dd]'1  AND json_extract_scalar(file_info, '$.source') = '[source]'; -- parsowane na każdym wierszu

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]'; -- typowana kolumna, czytana raz na plik

Zmieniły się dwie rzeczy. Filtr po metadanych pliku działa teraz na małej tabeli z jednym wierszem na plik, a nie na każdym wierszu dużej, i znika parsowanie JSON-a, które szło na każdym wierszu. Potem tych kilka pasujących plików jest łączonych z dużą tabelą, z małą tabelą po prawej stronie złączenia, tak jak zaleca AWS (AWS Big Data Blog).

Nazwane kroki

Długie zagnieżdżone zapytania zmieniły się w ciąg CTE, każde jako jeden nazwany krok, więc było jasne, który krok wykonuje kosztowną pracę.

Gdy zapytanie się nie powiedzie, proś o mniej

Nawet z poprawionym układem zapytanie na długim zakresie wciąż mogło dojść do limitu czasu. Zamiast ponawiać to samo zapytanie, zakres, który się nie powiódł, był dzielony na mniejsze, a części szły równolegle: miesiąc na cztery tygodnie, tydzień na dni, dzień na godziny. Łączna praca zostaje ta sama, ale każdy kawałek mieści się w limicie, a uruchamianie ich obok siebie skraca też czas od startu do końca. Symulacja poniżej wykonuje tę logikę na miesiącu danych (poglądowo, nie pomiar):

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); // jedno zapytanie na cały zakres  } catch (error) {2    if (range.unit === 'hour') throw error; // nie ma już nic mniejszego3    const parts = split(range, finer[range.unit]); // miesiąc staje się tygodniami, tydzień dniami, dzień godzinami    await Promise.all(parts.map(runRange)); // części idą równolegle  }}

Dzielona jest tylko ta część, która się nie powiedzie. Miesiąc, który przechodzi w jednym zapytaniu, zostaje jednym zapytaniem; tylko ciężki dzień w jego środku schodzi do godzin.

Co to zmieniło

Zapytania, które Athena wcześniej anulowała na limicie 240 minut, kończyły się wyraźnie przed nim.

Jeśli zadanie w Athenie jest wolne albo się nie udaje, sprawdzaj w tej kolejności: czy zapytanie filtruje po partycjach, czy parsuje albo liczy na każdym wierszu coś, co dałoby się zrobić raz, i czy filtruje małe tabele przed złączeniem z dużą. Gdy zakres wciąż jest za duży, dziel go po nieudanej próbie, zamiast ponawiać w całości.