Jak przechowywać minutowe odczyty ze 100 000 urządzeń
Spis treści8 ▾
Jeden odczyt na minutę z każdego urządzenia brzmi skromnie. To typowy wymóg, który trafia do specyfikacji, zanim ktokolwiek przemnoży liczby, a potem po cichu przesądza o architekturze wszystkiego, co znajduje się dalej w łańcuchu. Przez sporą część ostatnich lat pracowałem przy flotach liczących dziesiątki tysięcy urządzeń i w tej skali problemy sprowadzają się zwykle do arytmetyki, timingu i księgowości, byle robionych starannie.
Zacznij od arytmetyki
Sto tysięcy urządzeń raportujących co minutę daje średnio około 1667 wiadomości na sekundę, czyli 144 miliony wierszy dziennie i nieco ponad 52 miliardy rocznie. Żadna z tych liczb osobno nie przeraża nowoczesnej bazy danych, ale one się kumulują. Jeśli pojedynczy odczyt zawiera identyfikator urządzenia, znacznik czasu i cztery lub pięć niewielkich wartości z czujników, wiersz w Postgresie razem z nagłówkiem i indeksem klucza głównego zajmie mniej więcej sto bajtów. Po przemnożeniu wychodzi około 14 GB dziennie, czyli w okolicach 5 TB surowej historii rocznie, a to jeszcze bez replik i backupów.
Średnia jest zresztą najmniej ciekawą liczbą w tym zestawieniu. Przy wymiarowaniu liczy się szczyt, a o szczycie decyduje coś dużo mniej oczywistego niż wielkość floty.
Wszyscy naraz o pełnej minucie
Urządzenia uwielbiają okrągłe liczby. Jeśli firmware budzi się na timerze wyrównanym do zegara ściennego, duża część floty będzie próbowała połączyć się i wysłać dane w tej samej sekundzie, co minutę, bez końca. Średnie 1667 wiadomości na sekundę zamienia się wtedy w skok rzędu dziesiątek tysięcy, po którym zapada pięćdziesiąt kilka sekund niemal zupełnej ciszy. Brokery, load balancery i pule połączeń do bazy wymiaruje się pod ten skok, a za tę pojemność płacisz także wtedy, gdy panuje cisza.
Poprawka nie kosztuje prawie nic, jeśli zrobisz ją wcześnie. Każde urządzenie dostaje stałe przesunięcie w obrębie minuty, wyliczone z czegoś stabilnego, na przykład z hasha numeru seryjnego. Próbkuje nadal o pełnej minucie, ale wysyła dane w swoim momencie, więc publikacje rozkładają się na cały interwał. Odczyt dalej niesie znacznik czasu pomiaru, dlatego w reszcie systemu nic się nie zmienia, a krzywa obciążenia wypłaszcza się do postaci, pod którą da się planować. Zrobienie tego po wdrożeniu oznacza rollout OTA na wszystkie urządzenia, a to dokładnie ten rodzaj zmiany, którego wolisz nie robić pod presją.
Odbieranie wiadomości z brokera
Większość takich flot mówi po MQTT: z usługą zarządzaną w rodzaju AWS IoT Core albo z brokerem utrzymywanym we własnym zakresie, jak EMQX czy HiveMQ. Broker świetnie radzi sobie z utrzymywaniem stu tysięcy długo żyjących połączeń i rozsyłaniem wiadomości. Nie jest to jednak dobre miejsce na cokolwiek powolnego, więc pierwsze zadanie backendu to jak najszybciej odebrać od niego wiadomości i przenieść je w trwałe miejsce.
Po drodze lubię wstawić log, zwykle Kafkę albo Kinesis, zależnie od tego, gdzie działa reszta systemu. Konsumenci zapisujący do bazy mogą wtedy zostawać w tyle, restartować się albo przechodzić ponowne wdrożenie, a broker nawet tego nie zauważy. Do tego możesz odtworzyć ruch z całego dnia, kiedy nieuchronnie trafisz na błąd w parsowaniu któregoś pola. Dzięki shared subscriptions w MQTT można uruchomić kilka instancji mostka, które dzielą między siebie przychodzący strumień, więc ten pierwszy etap skaluje się poziomo i jest po prostu nudny.
Log partycjonuj po identyfikatorze urządzenia. Odczyty każdego urządzenia zachowują wtedy kolejność w obrębie partycji, a to bardzo upraszcza późniejszą deduplikację i wszelką logikę dotyczącą pojedynczych urządzeń.
Zapisuj wszystko batchami
Największy błąd wydajnościowy, jaki widuję, to zapisywanie osobnego wiersza dla każdej wiadomości. Przy 1667 insertach na sekundę baza spędzi większość czasu na narzucie transakcji, a bardzo mało na samym przechowywaniu danych. Zbierz w konsumencie kilka tysięcy wierszy i zapisz je jednym poleceniem albo skorzystaj z mechanizmu zapisu masowego, który oferuje baza, na przykład z COPY w Postgresie. ClickHouse jest pod tym względem jeszcze bardziej wymagający i naprawdę chce dostawać naraz batche liczące od tysięcy do dziesiątek tysięcy wierszy. Jeśli procesy zapisujące nie potrafią batchować po swojej stronie, buforowaniem po stronie serwera zajmą się asynchroniczne inserty ClickHouse'a.
Batch co sekundę czy dwie z każdego konsumenta w zupełności wystarczy. Dodatkowego opóźnienia nie zauważy nikt, kto patrzy na dashboard, a baza ci podziękuje.
Duplikaty i spóźnialskie odczyty
MQTT z QoS 1 gwarantuje dostarczenie wiadomości co najmniej raz, co w praktyce oznacza, że czasem dwa razy. Urządzenia, które tracą łączność, buforują poza tym odczyty i wysyłają je później: czasem po kilku godzinach, czasem po restarcie, po którym nie wiedzą już, które wiadomości zostały potwierdzone. Jeśli storage traktuje każdą przychodzącą wiadomość jako nowy fakt, średnie zaczną dryfować, a raporty rozliczeniowe, o ile je masz, będą błędne w sposób bardzo trudny do wychwycenia.
Najczystsze rozwiązanie to idempotentny zapis. Użyj identyfikatora urządzenia i znacznika czasu pomiaru jako klucza i pozwól bazie ignorować wiersze, które już widziała. Insert kosztuje przez to trochę więcej, za to nigdy nie musisz pisać joba deduplikującego.
Gdzie trzymać dane
Przy danych o takim charakterze w pierwszej kolejności sięgnąłbym po TimescaleDB albo ClickHouse. Oba wybory są dobre, a decyzja zwykle zależy od reszty stosu i od tego, co zespół już zna. TimescaleDB to Postgres, więc zachowujesz joiny z rejestrem urządzeń, transakcje i cały ekosystem narzędzi. ClickHouse jest szybszy przy dużych skanach analitycznych i bardzo oszczędnie obchodzi się z dyskiem, ale to osobny system z własną specyfiką operacyjną.
Schemat, który dobrze sprawdza się w TimescaleDB przy takim obciążeniu, wygląda mniej więcej tak:
CREATE TABLE readings (
device_id integer NOT NULL,
ts timestamptz NOT NULL,
temp real,
humidity real,
battery_mv smallint,
rssi smallint,
PRIMARY KEY (device_id, ts)
);
SELECT create_hypertable('readings', by_range('ts', INTERVAL '1 day'));
ALTER TABLE readings SET (
timescaledb.compress,
timescaledb.compress_segmentby = 'device_id',
timescaledb.compress_orderby = 'ts DESC'
);
SELECT add_compression_policy('readings', INTERVAL '2 days');
Dzienne chunki ograniczają każdą partycję do mniej więcej 144 milionów wierszy, które łatwo skompresować i tanio usunąć. Segmentowanie skompresowanych danych po urządzeniu i sortowanie ich po czasie układa odczyty z jednego urządzenia obok siebie, a wolno zmieniające się wartości z czujników kompresują się wtedy wyjątkowo dobrze. Oszczędności rzędu 90 procent lub więcej są tu normą, a telemetria z czujników to chyba najwdzięczniejsze do kompresji dane, jakie istnieją. Przy danych opisanych wyżej można realnie zejść z 5 TB rocznie do kilkuset gigabajtów.
Zapis sprowadza się wtedy do batchowego insertu, który po cichu pomija wszystko, co już jest w bazie:
INSERT INTO readings (device_id, ts, temp, humidity, battery_mv, rssi)
VALUES ($1, $2, $3, $4, $5, $6), ...
ON CONFLICT (device_id, ts) DO NOTHING;
Nikt nie czyta minutowych danych z zeszłego marca
Gdy dane już płyną, o kosztach decyduje pytanie, jak długo komukolwiek naprawdę potrzebny jest każdy poziom szczegółowości. Z mojego doświadczenia odpowiedź brzmi zwykle: krótko w przypadku surowych odczytów i długo w przypadku podsumowań. Inżynier, który debuguje urządzenie, chce danych minutowych z ostatnich kilku dni lub tygodni. Dashboardy produktowe i roczne raporty potrzebują agregatów godzinowych albo dziennych, a gdyby musiały skanować surowe wiersze, działałyby wolniej i kosztowały więcej.
Continuous aggregates w TimescaleDB albo materialized views w ClickHouse pozwalają aktualizować te podsumowania na bieżąco, w miarę napływu danych.
CREATE MATERIALIZED VIEW readings_hourly
WITH (timescaledb.continuous) AS
SELECT device_id,
time_bucket(INTERVAL '1 hour', ts) AS bucket,
avg(temp) AS temp_avg,
min(temp) AS temp_min,
max(temp) AS temp_max,
min(battery_mv) AS battery_min
FROM readings
GROUP BY device_id, bucket;
SELECT add_continuous_aggregate_policy('readings_hourly',
start_offset => INTERVAL '3 hours',
end_offset => INTERVAL '1 hour',
schedule_interval => INTERVAL '30 minutes');
SELECT add_retention_policy('readings', INTERVAL '90 days');
Przy takiej konfiguracji surowa tabela przechowuje zawsze tylko ostatnie trzy miesiące, widok godzinowy trzyma dowolnie długą historię za ułamek miejsca, a start offset rzędu kilku godzin daje spóźnionym odczytom szansę dotrzeć, zanim ich godzina zostanie domknięta.
Jeśli surowa historia może się kiedyś komuś przydać, na przykład przy sporze gwarancyjnym albo do trenowania modelu, eksportuj każdy dzień do Parquetu na object storage, zanim polityka retencji go usunie. Przechowywanie tam jest tanie, a narzędzia takie jak Athena czy DuckDB odpytują te pliki bezpośrednio, gdy raz na jakiś czas ktoś o nie poprosi.
Wszystko razem
Cały system nie jest skomplikowany. Urządzenia rozkładają publikacje na całą minutę, broker utrzymuje połączenia, log przyjmuje strumień, niewielka grupa konsumentów idempotentnie zapisuje batche do bazy szeregów czasowych, która agresywnie kompresuje i agreguje dane, a wszystko, co jest już na tyle stare, że zrobiło się nudne, trafia do object storage. Każdy z tych elementów jest dobrze znany.
Prawdziwego wysiłku wymaga świadome ustalenie tych wszystkich liczb, zanim flota powstanie, póki zmiany są jeszcze tanie. Jeśli planujesz coś w tej skali i przydałoby ci się drugie spojrzenie, zawsze chętnie porozmawiam.