DBA: właściwie organizujemy synchronizacje i importy

W przypadku złożonego przetwarzania dużych zbiorów danych (różne procesy ETL: importy, konwersje i synchronizacje z zewnętrznym źródłem) często pojawia się potrzeba tymczasowego "zapamiętania" i szybkiego przetworzenia czegoś obszernego.

Typowe zadanie tego rodzaju brzmi zazwyczaj mniej więcej tak: "Tu dział księgowości wyeksportował z klient-banku ostatnie otrzymane płatności, musimy je szybko załadować na stronę i powiązać z rachunkami"

Jednak gdy objętość tego "czegoś" zaczyna być mierzona w setkach megabajtów, a usługa musi nadal działać w trybie 24×7, pojawia się wiele skutków ubocznych, które będą przeszkadzać w pracy.

Aby sobie z nimi poradzić w PostgreSQL (i nie tylko tam), można wykorzystać pewne możliwości optymalizacji, które pozwolą przetwarzać wszystko szybciej i przy mniejszym zużyciu zasobów.

1. Gdzie ładować?

Na początku ustalmy, gdzie możemy załadować dane, które chcemy "przetworzyć".

1.1. Tabele tymczasowe (TEMPORARY TABLE)

W zasadzie dla PostgreSQL tabele tymczasowe to te same tabele jak wszystkie inne. Dlatego błędne są przesądy typu "wszystko tam jest przechowywane tylko w pamięci, a ona może się skończyć". Ale są również pewne istotne różnice.

Własna "przestrzeń nazw" dla każdego połączenia z bazą danych

Jeśli dwa połączenia spróbują jednocześnie wykonać CREATE TABLE x, to ktoś na pewno dostanie błąd unikalności obiektów bazy danych.

Jeśli jednak obie próby zostaną wykonane UTWÓRZ Tymczasowa Tabela x, to obie się powiodą, a każdy otrzyma swój egzemplarz tabeli. Nie będzie między nimi nic wspólnego.

"Autodestrukcja" przy rozłączeniu

Po zamknięciu połączenia wszystkie tabele tymczasowe są automatycznie usuwane, dlatego nie ma sensu wykonywać DROP TABLE x , oprócz...

Jeśli pracujesz przez pgbouncer w trybie transakcyjnym, to baza nadal uważa, że to połączenie jest aktywne, a w nim tabela tymczasowa nadal istnieje.

Dlatego próba ponownego jej utworzenia z innego połączenia do pgbouncer spowoduje błąd. Ale można to obejść, korzystając z UTWÓRZ TYMCHASOWĄ TABELĘ JEŚLI NIE ISTNIEJE x.

Jednak lepiej tego nie robić, bo później można "nagle" znaleźć dane pozostałe po "poprzednim właścicielu". Zamiast tego znacznie lepiej przeczytać dokumentację i zobaczyć, że przy tworzeniu tabeli istnieje możliwość dodania PRZY ZATWIERDZENIU DROP — czyli po zakończeniu transakcji tabela zostanie automatycznie usunięta.

Brak replikacji

Ze względu na to, że należy tylko do określonego połączenia, tabele tymczasowe nie są replikowane. Z kolei pozwala to uniknąć podwójnego zapisu danych w heap + WAL, dlatego INSERT/UPDATE/DELETE do niej jest znacznie szybsze.

Jednak ponieważ tabela tymczasowa to wciąż „prawie zwykła” tabela, nie można jej również utworzyć na replikacji. Przynajmniej na razie, chociaż odpowiednia łatka jest już od dawna przygotowywana.

1.2. Tabele niezarejestrowane (UNLOGGED TABLE)

Ale co zrobić na przykład, jeśli masz jakiś ciężki proces ETL, którego nie da się zrealizować w ramach jednej transakcji, a jednak masz pgbouncer w trybie transakcyjnym?..

Lub strumień danych jest na tyle duży, że nie wystarcza przepustowości jednego połączenia z bazą danych (czytaj, jednego procesu na CPU)?..

Lub część operacji przebiega asynchronicznie w różnych połączeniach?..

Jest tylko jedna opcja — tymczasowo stworzyć tabelę zwykłą. Kalambur, tak. To znaczy:

• stwórz „swoje” tabele z maksymalnie losowymi nazwami, aby się z nikim nie zderzyć
• Ekspert: załadowałem w nie dane zewnętrznego źródła
• Transformuj: przekształciłem, wypełniłem kluczowe powiązane pola
• Ładowanie: załadowałem gotowe dane do docelowych tabel
• usunąłem „swoje” tabele

A teraz — łyżka dziegciu. Właściwie, cały zapis w PostgreSQL odbywa się dwukrotnie — najpierw w WAL, a następnie w ciała tabel/indeksów. Wszystko to zostało zrobione w celu wsparcia ACID i poprawnej widoczności danych między COMMITtransakcjami 'wnętrznymi' i ROLLBACKtransakcjami 'zewnętrznymi'.

Ale to nie jest nam potrzebne! Cały proces przeszedł całkowicie pomyślnie lub nie.Nie ma znaczenia, ile w nim będzie pośrednich transakcji — nie interesuje nas „kontynuowanie procesu z połowy”, zwłaszcza kiedy nie wiadomo, gdzie to było.

W tym celu deweloperzy PostgreSQL już w wersji 9.1 wprowadzili coś takiego jak tabele niezarejestrowane (UNLOGGED):

Z tym wskazaniem tabela jest tworzona jako niezarejestrowana. Dane zapisywane w tabelach niezarejestrowanych nie przechodzą przez dziennik przedzapisowy (patrz rozdział 29), w wyniku czego takie tabele działają znacznie szybciej niż zwykłe. Jednak nie są chronione przed awarią; w przypadku awarii lub nagłego wyłączenia serwera tabela niezarejestrowana zostaje automatycznie ucięta.. Ponadto, zawartość niezarejestrowanej tabeli nie jest replikowana. na serwery podrzędne. Wszelkie indeksy tworzone dla tabeli niezapisanej w dzienniku stają się automatycznie niezapisane w dzienniku.

Krótko mówiąc, będzie znacznie szybsze, ale jeśli serwer bazy danych „padnie” — to będzie niewygodne. Ale jak często to się zdarza i czy Twój proces ETL umie to poprawnie naprawić „z połowy” po „ożywieniu” bazy danych?..

Jeśli nie, a powyższy przypadek przypomina Twój — użyj UNLOGGED, ale nigdy nie włączaj tego atrybutu w rzeczywistych tabelach, z których dane są dla Ciebie cenne.

1.3. ON COMMIT { DELETE ROWS | DROP }

Ta konstrukcja umożliwia podczas tworzenia tabeli określenie automatycznego zachowania po zakończeniu transakcji.

O PRZY ZATWIERDZENIU DROP Już pisałem wyżej, generuje DROP TABLE, ale w przypadku PRZY ZATWIERDZENIU USUŃ WIERSZE sytuacja jest ciekawsza — tutaj generowane jest TRUNCATE TABLE.

Ponieważ cała infrastruktura przechowywania metadanych tabeli tymczasowej jest dokładnie taka sama, jak w przypadku zwykłej, to ciągłe tworzenie i usuwanie tabel tymczasowych prowadzi do znacznego „rozrostu” tabel systemowych pg_class, pg_attribute, pg_attrdef, pg_depend,…

Wyobraź sobie teraz, że masz robota na bezpośrednim połączeniu z bazą danych, który co sekundę otwiera nową transakcję, tworzy, wypełnia, przetwarza i usuwa tymczasową tabelę… Śmieci w tabelach systemowych nagromadzi się w nadmiarze, a to powoduje dodatkowe opóźnienia przy każdej operacji.

Ogólnie rzecz biorąc, nie rób tak! W takim przypadku znacznie skuteczniej CREATE TEMPORARY TABLE x ... ON COMMIT DELETE ROWS wyciągnąć poza cykl transakcji — wtedy na początku każdej nowej transakcji tabele już będą istnieć (oszczędzamy wywołanie UTWÓRZ), ale będą puste, dzięki TRUNCATE (jego wywołanie też zaoszczędziliśmy) po zakończeniu poprzedniej transakcji.

1.4. LIKE… INCLUDING …

Wspomniałem na początku, że jednym z typowych przypadków użycia tabel tymczasowych są różnego rodzaju importy — a programista zmęczony kopiowaniem i wklejaniem listy pól tabeli docelowej w deklaracji swojej tabeli tymczasowej…

Ale lenistwo jest motorem postępu! Dlatego utworzenie nowej tabeli „na podstawie” można wykonać znacznie łatwiej:

CREATE TEMPORARY TABLE import_table(
  LIKE target_table
);

Ponieważ później można wlać do tej tabeli naprawdę dużo danych, przeszukiwanie jej stanie się niezbyt szybkie. Ale jest na to tradycyjne rozwiązanie — indeksy! I tak, tabela tymczasowa również może mieć indeksy.

Ponieważ często wymagane indeksy pokrywają się z indeksami tabeli docelowej, można po prostu napisać LIKE target_table W tym indeksy.

Jeśli potrzebujesz również DOMYŚLNY-wartości (np. do wypełnienia wartości klucza podstawowego), można skorzystać LIKE target_table W TYM DOMYŚLNE. Albo po prostu — LIKE target_table W TYM WSZYSTKO — skopiuje domyślne wartości, indeksy, constraints,…

Ale tu już trzeba zrozumieć, że jeśli tworzyłeś import-tabelę od razu z indeksami, to ładowanie danych będzie trwało dłużej, niż jeśli najpierw wszystko załadujesz, a potem dorobisz indeksy — zobacz jako przykład, jak to robi pg_dump.

Ogólnie, RTFM!

2. Jak pisać?

Powiem krótko — używaj COPY-strumienia zamiast „partii” INSERT, przyspieszenie w dziesiątkach razy. Można nawet bezpośrednio z wcześniej przygotowanego pliku.

3. Jak przetwarzać?

Zatem, niech nasz przypadek wygląda mniej więcej tak:

• masz w bazie tabelę z danymi klientów na 1M rekordów
• każdego dnia klient przesyła Ci nowy pełny „obraz”
• z doświadczenia wiesz, że z każdym razem zmienia się nie więcej niż 10K rekordów

Klasycznym przykładem takiej sytuacji jest baza KŁADR — jest ogromna liczba adresów, ale w każdej cotygodniowej wyeksportowanej wersji zmian (zmiany nazw miejscowości, łączenie ulic, pojawianie się nowych domów) jest zaledwie kilka nawet na skalę całego kraju.

3.1. Algorytm pełnej synchronizacji

Dla uproszczenia przyjmij, że nie musisz nawet restrukturyzować danych — wystarczy przekształcić tabelę do odpowiedniego formatu, to znaczy:

• usunąć wszystko, czego już nie ma
• aktualizować wszystko, co już było i należy zaktualizować
• wstawić wszystko, czego jeszcze nie było

Dlaczego właśnie w takiej kolejności należy wykonywać operacje? Ponieważ w ten sposób rozmiar tabeli wzrośnie minimalnie (pamiętaj o MVCC!).

DELETE FROM dst

Nie, oczywiście można się obejść tylko dwoma operacjami:

• usunąć (USUŃ) w zasadzie wszystkie
• wstawić wszystkie z nowego obrazu

Ale przy tym, dzięki MVCC, rozmiar tabeli zwiększy się dokładnie dwukrotnie! Uzyskać +1M obrazów rekordów w tabeli z powodu aktualizacji 10K — to dość marnotrawstwo…

TRUNCATE dst

Bardziej doświadczony programista wie, że cały cały stół można wystarczająco tanio wyczyścić:

• wyczyszczenia (TRUNCATE) tabelę całkowicie
• wstawić wszystkie z nowego obrazu

Metoda skuteczna, czasami całkiem zastosowalna, ale jest pewien problem… Wlewanie 1M rekordów zajmie nam sporo czasu, więc nie możemy pozwolić sobie na umieszczenie tabeli w pustej formie przez cały ten czas (jak by się stało bez opakowania w pojedynczą transakcję).

A zatem:

• rozpoczynamy długą transakcję
• TRUNCATE nakłada AccessExclusive-blokadę
• długo robimy wstawki, a wszyscy inni w tym czasie nawet nie mogą SELECT

Coś nie tak wychodzi…

ALTER TABLE… RENAME… / DROP TABLE …

Można na przykład wgrać wszystko do osobnej nowej tabeli, a następnie po prostu zmienić jej nazwę na starą. Kilka drobnych, uciążliwych rzeczy:

• także AccessExclusive, chociaż znacznie mniej czasochłonnie
• wszystkie plany zapytań/statystyki tej tabeli zostaną zresetowane, należy uruchomić ANALYZE
• wszystkie klucze obce (FK) do tabeli

Był WIP-patch od Simona Riggsa, który proponował ALTER-operację do podmiany treści tabeli na poziomie plików, bez dotykania statystyk i FK, ale nie uzyskał wymaganego kworum.

DELETE, UPDATE, INSERT

Więc decydujemy się na nieblokującą opcję z trzech operacji. Prawie trzech… Jak to zrobić w najbardziej efektywny sposób?

-- wszystko robimy w ramach transakcji, aby nikt nie widział "pośrednich" stanów
BEGIN;

-- tworzymy tymczasową tabelę z importowanymi danymi
CREATE TEMPORARY TABLE tmp(
  LIKE dst INCLUDING INDEXES -- według wzoru, łącznie z indeksami
) ON COMMIT DROP; -- po zakończeniu transakcji nie jest nam potrzebna

-- szybko wlewamy nową wersję przez COPY
COPY tmp FROM STDIN;
-- ...
-- .

-- usuwamy brakujące
DELETE FROM
  dst D
USING
  dst X
LEFT JOIN
  tmp Y
    USING(pk1, pk2) -- pola klucza głównego
WHERE
  (D.pk1, D.pk2) = (X.pk1, X.pk2) AND
  Y IS NOT DISTINCT FROM NULL; -- "antijoin"

-- aktualizujemy pozostałe
UPDATE
  dst D
SET
  (f1, f2, f3) = (T.f1, T.f2, T.f3)
FROM
  tmp T
WHERE
  (D.pk1, D.pk2) = (T.pk1, T.pk2) AND
  (D.f1, D.f2, D.f3) IS DISTINCT FROM (T.f1, T.f2, T.f3); -- nie ma potrzeby aktualizowania zgodnych

-- wstawiamy brakujące
INSERT INTO
  dst
SELECT
  T.*
FROM
  tmp T
LEFT JOIN
  dst D
    USING(pk1, pk2)
WHERE
  D IS NOT DISTINCT FROM NULL;

COMMIT;

3.2. Postprocessing importu

W tym samym KLADe, wszystkie zmienione rekordy należy dodatkowo przepuścić przez postprocessing — znormalizować, wydobyć słowa kluczowe, dostosować do potrzebnych struktur. Ale jak dowiedzieć się — co dokładnie się zmieniało, nie komplikując przy tym kodu synchronizacji, najlepiej w ogóle go nie dotykając?

Jeśli w momencie synchronizacji tylko twoje procesy mają dostęp do zapisu, można skorzystać z triggera, który zbierze dla nas wszystkie zmiany:

-- tabele docelowe
CREATE TABLE kladr(...);
CREATE TABLE kladr_house(...);

-- tabele z historią zmian
CREATE TABLE kladr$log(
  ro kladr, -- tutaj znajdują się pełne obrazy rekordów starych/nowych
  rn kladr
);

CREATE TABLE kladr_house$log(
  ro kladr_house,
  rn kladr_house
);

-- ogólna funkcja logowania zmian
CREATE OR REPLACE FUNCTION diff$log() RETURNS trigger AS $$
DECLARE
  dst varchar = TG_TABLE_NAME || '$log';
  stmt text = '';
BEGIN
  -- sprawdzamy konieczność logowania przy aktualizacji rekordu
  IF TG_OP = 'UPDATE' THEN
    IF NEW IS NOT DISTINCT FROM OLD THEN
      RETURN NEW;
    END IF;
  END IF;
  -- tworzymy rekord logu
  stmt = 'INSERT INTO ' || dst::text || '(ro,rn)VALUES(';
  CASE TG_OP
    WHEN 'INSERT' THEN
      EXECUTE stmt || 'NULL,$1)' USING NEW;
    WHEN 'UPDATE' THEN
      EXECUTE stmt || '$1,$2)' USING OLD, NEW;
    WHEN 'DELETE' THEN
      EXECUTE stmt || '$1,NULL)' USING OLD;
  END CASE;
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

Teraz możemy nałożyć (lub włączyć przez) wyzwalacze przed rozpoczęciem synchronizacji ALTER TABLE ... ENABLE TRIGGER ...):

CREATE TRIGGER log
  AFTER INSERT OR UPDATE OR DELETE
  ON kladr
    FOR EACH ROW
      EXECUTE PROCEDURE diff$log();

CREATE TRIGGER log
  AFTER INSERT OR UPDATE OR DELETE
  ON kladr_house
    FOR EACH ROW
      EXECUTE PROCEDURE diff$log();

A potem spokojnie wyciągamy wszystkie potrzebne nam zmiany z tabel log, a następnie uruchamiamy dodatkowe przetwarzacze.

3.3. Import powiązanych zestawów

Powyżej omówiliśmy przypadki, w których struktury danych źródła i odbiorcy są zgodne. Ale co zrobić, jeśli eksport z zewnętrznego systemu ma format różniący się od struktury przechowywania w naszej bazie?

Weźmy za przykład przechowywanie klientów i związanych z nimi faktur, klasyczny przypadek „wiele-do-jednego”:

CREATE TABLE client(
  client_id
    serial
      PRIMARY KEY
, inn
    varchar
      UNIQUE
, name
    varchar
);

CREATE TABLE invoice(
  invoice_id
    serial
      PRIMARY KEY
, client_id
    integer
      REFERENCES client(client_id)
, number
    varchar
, dt
    date
, sum
    numeric(32,2)
);

A oto eksport z zewnętrznego źródła przychodzi do nas w formie „wszystko w jednym”:

CREATE TEMPORARY TABLE invoice_import(
  client_inn
    varchar
, client_name
    varchar
, invoice_number
    varchar
, invoice_dt
    date
, invoice_sum
    numeric(32,2)
);

Oczywiście dane dotyczące klientów mogą być zdublowane w takim wariancie, a podstawowym rekordem jest „faktura”:

0123456789;Wania;A-01;2020-03-16;1000.00
9876543210;Petia;A-02;2020-03-16;666.00
0123456789;Wania;B-03;2020-03-16;9999.00

Dla modelu po prostu wstawimy nasze dane testowe, ale pamiętajmy — COPY efektywniej!

INSERT INTO invoice_import
VALUES
  ('0123456789', 'Wania', 'A-01', '2020-03-16', 1000.00)
, ('9876543210', 'Petia', 'A-02', '2020-03-16', 666.00)
, ('0123456789', 'Wania', 'B-03', '2020-03-16', 9999.00);

Najpierw wyodrębnijmy te „kategorie”, na które nasze „fakty” się odnoszą. W naszym przypadku faktury odnoszą się do klientów:

UTWÓRZ TYMCZASOWĄ TABELĘ client_import AS
WYBIERZ UNIKATOWE NA(client_inn)
-- można po prostu WYBRAĆ UNIKATOWE, jeśli dane są z góry nieprzeczące
  client_inn inn
, client_name "name"
Z FROM
  invoice_import;

Aby poprawnie powiązać faktury z identyfikatorami klientów, musimy najpierw poznać lub wygenerować te identyfikatory. Dodamy dla nich pola:

ALTER TABLE invoice_import DODAJ KOLUMNĘ client_id integer;
ALTER TABLE client_import DODAJ KOLUMNĘ client_id integer;

Skorzystamy z opisanego powyżej sposobu synchronizacji tabel z niewielką poprawką — nie będziemy nic aktualizować ani usuwać w docelowej tabeli, ponieważ import klientów jest dla nas "tylko do dodawania":

-- wstawiamy w tabeli importu ID już istniejących rekordów
UPDATE
  client_import T
USTAW
  client_id = D.client_id
Z
  client D
GDZIE
  T.inn = D.inn; -- unikalny klucz

-- wstawiamy brakujące rekordy i ustawiamy ich ID
WITH ins AS (
  WSTAW DO client(
    inn
  , name
  )
  WYBIERZ
    inn
  , name
  Z
    client_import
  GDZIE
    client_id IS NULL -- jeśli ID nie było ustawione
  ZWRÓĆ *
)
UPDATE
  client_import T
USTAW
  client_id = D.client_id
Z
  ins D
GDZIE
  T.inn = D.inn; -- unikalny klucz

-- ustawiamy ID klientów dla rekordów faktur
UPDATE
  invoice_import T
USTAW
  client_id = D.client_id
Z
  client_import D
GDZIE
  T.client_inn = D.inn; -- klucz aplikacyjny

Właściwie to wszystko — w invoice_import teraz mamy wypełnione pole powiązania client_id, z którym wstawimy fakturę.

Źródło: habr.com