DBA: organisera synkroniseringar och importer på ett kompetent sätt

Vid komplex bearbetning av stora datamängder (olika ETL-processer: importer, konverteringar och synkronisering med en extern källa) finns det ofta ett behov tillfälligt "kom ihåg" och omedelbart snabbt bearbeta något omfattande.

En typisk uppgift av det här slaget låter vanligtvis ungefär så här: "Här bokföring laddas ner från klient-bank de senaste mottagna betalningarna måste de snabbt laddas upp till webbplatsen och länkas till kontona"

Men när volymen av detta "något" börjar mätas i hundratals megabyte, och tjänsten måste fortsätta att arbeta med databasen i 24×7-läge, uppstår många biverkningar som kommer att förstöra ditt liv.

För att hantera dem i PostgreSQL (och inte bara i det), kan du använda några optimeringsfunktioner som gör att du kan bearbeta allt snabbare och med mindre resursförbrukning.

1. Vart ska man skicka?

Låt oss först bestämma var vi kan ladda upp den data vi vill "behandla".

1.1. Tillfälliga tabeller (TEMPORARY TABLE)

I princip, för PostgreSQL, är temporära tabeller desamma som alla andra. Därför vidskepelse som t.ex "allt lagras bara i minnet, och det kan ta slut". Men det finns också några betydande skillnader.

Eget "namnområde" för varje anslutning till databasen

Om två anslutningar görs samtidigt CREATE TABLE x, då kommer någon definitivt att få det icke-unika fel DB-objekt.

Men om båda försöker uppfylla CREATE TEMPORARY TABLE x, då kommer båda att göra det normalt, och alla kommer att få det ditt exemplar tabeller. Och det kommer inte att finnas något gemensamt mellan dem.

"Självförstörande" vid frånkoppling

När anslutningen stängs släpps alla temporära tabeller automatiskt, så det är inte nödvändigt att utföra "manuellt" DROP TABLE x det är ingen mening, förutom...

Om du jobbar igenom pgbouncer i transaktionsläge, fortsätter databasen att anse att den här anslutningen fortfarande är aktiv och att den här temporära tabellen fortfarande finns i den.

Om du försöker skapa den igen, från en annan anslutning till pgbouncer, kommer det därför att resultera i ett fel. Men detta kan kringgås genom att använda CREATE TEMPORARY TABLE IF NOT EXISTS x.

Det är dock bättre att inte göra detta, för då kan du "plötsligt" upptäcka data som blivit över från den "föregående ägaren". Istället är det mycket bättre att läsa manualen och se att när man skapar en tabell finns det en möjlighet att lägga till ON COMMIT DROP - det vill säga när transaktionen är slutförd kommer tabellen automatiskt att raderas.

Icke-replikering

Eftersom de är anslutningsspecifika replikeras inte temporära tabeller. Men Detta eliminerar behovet av dubbelinmatning i heap + WAL, så INSERT/UPDATE/DELETE i den är betydligt snabbare.

Men eftersom en temporär tabell fortfarande är en "nästan vanlig" tabell, kan den inte heller skapas på en replik. Åtminstone för nu, även om motsvarande patch har funnits länge.

1.2. OLOGGAD BORD

Men vad ska du göra, till exempel om du har en besvärlig ETL-process som inte kan implementeras inom en enda transaktion, och du fortfarande har pgbouncer i transaktionsläge? ..

Eller så är dataflödet så stort att inte tillräckligt med bandbredd för en anslutning med en DB (läs, en process per CPU)?..

Eller så pågår en del av operationerna asynkront i olika sammanhang?...

Det finns bara ett alternativ här - skapa tillfälligt en icke-tillfällig tabell. En ordlek, ja. Som är:

• skapade "hans" tabeller med maximalt slumpmässiga namn för att inte korsa vägar med någon
• Utdrag: fyllde dem med data från en extern källa
• Förvandla: konverterade, ifyllda nyckelbindningsfält
• Ladda: överförde färdiga data till måltabellerna
• raderade "mina" tabeller

Och nu - en fluga i glädjen. I grund och botten, alla skrivningar i PostgreSQL händer två gånger - först i WAL, sedan in i tabell-/indexkropparna. Allt detta görs för att stödja ACID och korrekt datasynlighet mellan COMMIT'nötlig och ROLLBACK'nötliga transaktioner.

Men det behöver vi inte! Vi har hela processen antingen godkänts helt framgångsrikt eller inte. Det spelar ingen roll hur många mellanliggande transaktioner det finns – vi är inte intresserade av att "fortsätta processen från mitten", särskilt när det inte är klart var det var.

För detta ändamål introducerade PostgreSQL-utvecklare en sådan sak tillbaka i version 9.1 ologgade tabeller:

Med denna specifikation skapas tabellen som ologgad. Data som skrivs till ologgade tabeller går inte igenom framskrivningsloggen (se kapitel 29), vilket gör att sådana tabeller arbeta mycket snabbare än vanligt. Men de är inte immuna mot misslyckanden; i händelse av serverfel eller nödavstängning ologgad tabell automatiskt trunkerad. Dessutom innehållet i den ologgade tabellen replikerar inte till slavservrar. Alla index som skapas på en ologgad tabell görs automatiskt ologgade.

Kort sagt, det kommer att gå mycket snabbare, men om databasservern "faller" - blir det obehagligt. Men hur ofta händer detta, och kan din ETL-process modifiera den korrekt "från mitten" efter "återupplivandet" av databasen?

Om inte, och fallet ovan liknar ditt, använd UNLOGGED, men aldrig aktivera inte detta attribut på riktiga tabeller, vars data ligger dig varmt om hjärtat.

1.3. ON COMMIT { DELETE ROWS | DROP }

Denna konstruktion låter dig specificera automatiskt beteende när en transaktion är slutförd när du skapar en tabell.

Про ON COMMIT DROP Jag skrev redan ovan, det genererar DROP TABLE, men med ON COMMIT DELETE ROWS situationen är mer intressant - här genereras den TRUNCATE TABLE.

Eftersom hela infrastrukturen för att lagra metabeskrivningen för en temporär tabell är exakt densamma som för en vanlig tabell, ständigt skapande och radering av tillfälliga tabeller leder till kraftig "svällning" av systemtabeller pg_class, pg_attribute, pg_attrdef, pg_depend,...

Föreställ dig nu att du har en arbetare på en direktanslutning till DB, som öppnar en ny transaktion varje sekund, skapar, fyller, bearbetar och tar bort en temporär tabell... Skräp i systemtabellerna kommer att ackumuleras i överskott, och detta är en onödig nedgång för varje operation.

I allmänhet, gör inte detta! I det här fallet är det mycket mer effektivt CREATE TEMPORARY TABLE x ... ON COMMIT DELETE ROWS ta den ur transaktionscykeln - sedan är tabellerna redan i början av varje ny transaktion kommer att finnas (sparar ett samtal CREATE), Men kommer att vara tom, tack vare TRUNCATE (vi sparade också samtalet) när vi slutförde den föregående transaktionen.

1.4. GILLA…INKLUSIVE…

Jag nämnde i början att ett av de typiska användningsfallen för temporära tabeller är olika typer av import - och utvecklaren kopierar trött och klistrar in listan med fält i måltabellen i deklarationen av sin tillfälliga...

Men lathet är motorn för framsteg! Det är därför skapa en ny tabell "efter prov" det kan göras mycket enklare:

CREATE TEMPORARY TABLE import_table(
  LIKE target_table
);

Eftersom det är möjligt att generera ganska mycket data i den här tabellen senare kommer det inte att gå snabbt att söka igenom den alls. Men det finns en traditionell lösning på detta - index! Och ja, en temporär tabell kan också ha index.

Eftersom de nödvändiga indexen ofta sammanfaller med måltabellens index kan du helt enkelt skriva LIKE target_table INCLUDING INDEXES.

Om du också behöver DEFAULT- värden (till exempel för att fylla i värden för primärnyckeln) kan du använda LIKE target_table INCLUDING DEFAULTS. Eller helt enkelt - LIKE target_table INCLUDING ALL - kopierar standardvärden, index, begränsningar,...

Men här måste du förstå att om du skapade importera tabell med index omedelbart, då tar data längre tid att fylla, än om du först laddar upp allt och sedan rullar ut indexen - titta på hur den gör detta som ett exempel pg_dump.

I allmänhet, RTFM!

2. Hur skriver man?

Jag säger bara det - använd det COPY-stream istället för "packa" INSERT, acceleration flera gånger. Du kan till och med göra det direkt från en förgenererad fil.

3. Hur bearbetar man?

Så låt oss säga att vårt intro ser ut ungefär så här:

• Du har en tabell med kunddata lagrad i din databas 1 miljon rekord
• Varje dag skickar kunden en ny till dig fullständig "bild"
• av erfarenhet vet du det då och då inte mer än 10 XNUMX poster ändras

Ett klassiskt exempel på en sådan situation är KLADR bas — det finns många adresser, men i varje veckouppladdning sker mycket få ändringar (byte av bebyggelse, sammanslagning av gator, utseende av nya hus) även i nationell skala.

3.1. Fullständig synkroniseringsalgoritm

För enkelhetens skull, låt oss anta att du inte ens behöver omstrukturera data - bara ta tabellen till önskat format, det vill säga:

• ta bort allt som inte finns längre
• uppdatering allt som redan fanns behöver uppdateras
• infoga allt som inte har hänt än

Varför är det nödvändigt att utföra operationer i just denna ordning? Eftersom det är det enda sättet att bordsstorleken kommer att växa minimalt (kom ihåg MVCC!).

DELETE FRÅN dst

Nej, självklart klarar du dig med bara två operationer:

• ta bort (DELETE) allt i allmänhet
• infoga allt från den nya bilden

Men samtidigt, tack vare MVCC, bordsstorleken kommer att öka exakt två gånger! Att få +1M tabellpostbilder på grund av 10K-uppdatering är inte så mycket redundans...

TRUNCATE dst

En mer erfaren utvecklare vet att hela bordet kan städas ganska billigt:

• ren (TRUNCATE) hela bordet
• infoga allt från den nya bilden

Metoden är effektiv, ibland ganska tillämplig, men det finns ett problem... Vi kommer att hälla ut 1 miljon poster under en lång tid, så vi har inte råd att lämna bordet tomt under hela denna tid (vilket kommer att hända utan att slå in det i en enda transaktion).

Vilket betyder:

• vi börjar lång transaktion
• TRUNCATE ålägger AccessExklusiv-blockering
• vi tar lång tid att sätta in, och alla andra samtidigt de kan inte ens SELECT

Något går inte bra...

ÄNDRA TABELL... BYT DAMN... / SLIPP TABELL...

Ett annat alternativ är att ladda upp allt till en separat ny tabell och sedan helt enkelt byta namn på den i stället för den gamla. Ett par otäcka småsaker:

• samma också AccessExklusiv, men betydligt mindre i tid
• alla frågeplaner/statistik för denna tabell återställs, måste köra ANALYSE
• alla externa nycklar är trasiga (FK) på bordet

Det fanns en WIP-patch från Simon Riggs som föreslog att man skulle göra det ALTER- en operation för att ersätta tabellkroppen på filnivå, utan att röra statistiken och FK, men samlade inte in ett beslutfört.

DELETE, UPPDATERA, INFOGA

Så vi kommer att nöja oss med en icke-blockerande variant av tre operationer. Nästan tre... Hur gör man detta mest effektivt?

-- все делаем в рамках транзакции, чтобы никто не видел "промежуточных" состояний
BEGIN;

-- создаем временную таблицу с импортируемыми данными
CREATE TEMPORARY TABLE tmp(
  LIKE dst INCLUDING INDEXES -- по образу и подобию, вместе с индексами
) ON COMMIT DROP; -- за рамками транзакции она нам не нужна

-- быстро-быстро вливаем новый образ через COPY
COPY tmp FROM STDIN;
-- ...
-- .

-- удаляем отсутствующие
DELETE FROM
  dst D
USING
  dst X
LEFT JOIN
  tmp Y
    USING(pk1, pk2) -- поля первичного ключа
WHERE
  (D.pk1, D.pk2) = (X.pk1, X.pk2) AND
  Y IS NOT DISTINCT FROM NULL; -- "антиджойн"

-- обновляем оставшиеся
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); -- незачем обновлять совпадающие

-- вставляем отсутствующие
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. Efterbearbetning av import

I samma KLADR måste alla ändrade poster dessutom köras genom efterbearbetning - normaliseras, nyckelord markeras och föras till de strukturer som krävs. Men hur vet du det? exakt vad som förändrades, utan att komplicera synkroniseringskoden, helst utan att röra den alls?

Om endast din process har skrivåtkomst vid tidpunkten för synkroniseringen kan du använda en trigger som samlar in alla ändringar åt oss:

-- целевые таблицы
CREATE TABLE kladr(...);
CREATE TABLE kladr_house(...);

-- таблицы с историей изменений
CREATE TABLE kladr$log(
  ro kladr, -- тут лежат целые образы записей старой/новой
  rn kladr
);

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

-- общая функция логирования изменений
CREATE OR REPLACE FUNCTION diff$log() RETURNS trigger AS $$
DECLARE
  dst varchar = TG_TABLE_NAME || '$log';
  stmt text = '';
BEGIN
  -- проверяем необходимость логгирования при обновлении записи
  IF TG_OP = 'UPDATE' THEN
    IF NEW IS NOT DISTINCT FROM OLD THEN
      RETURN NEW;
    END IF;
  END IF;
  -- создаем запись лога
  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;

Nu kan vi tillämpa triggers innan synkroniseringen startar (eller aktivera dem via 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();

Och sedan extraherar vi lugnt alla ändringar vi behöver från loggtabellerna och kör dem genom ytterligare hanterare.

3.3. Importera relaterade uppsättningar

Ovan övervägde vi fall där datastrukturerna för källan och mottagaren är desamma. Men vad ska man göra om nedladdningen från ett externt system har ett annat format än lagringsstrukturen i vår databas?

Låt oss ta som exempel lagringen av kunder och deras fakturor, ett klassiskt "många-till-en"-alternativ:

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)
);

Men nedladdningen från en extern källa kommer till oss i form av "allt i ett":

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

Uppenbarligen kan kunddata dupliceras i den här versionen, och huvudposten är "kontot":

0123456789;Вася;A-01;2020-03-16;1000.00
9876543210;Петя;A-02;2020-03-16;666.00
0123456789;Вася;B-03;2020-03-16;9999.00

För modellen kommer vi bara att infoga våra testdata, men kom ihåg - COPY effektivare!

INSERT INTO invoice_import
VALUES
  ('0123456789', 'Вася', 'A-01', '2020-03-16', 1000.00)
, ('9876543210', 'Петя', 'A-02', '2020-03-16', 666.00)
, ('0123456789', 'Вася', 'B-03', '2020-03-16', 9999.00);

Låt oss först lyfta fram de "avsnitt" som våra "fakta" hänvisar till. I vårt fall avser fakturorna kunder:

CREATE TEMPORARY TABLE client_import AS
SELECT DISTINCT ON(client_inn)
-- можно просто SELECT DISTINCT, если данные заведомо непротиворечивы
  client_inn inn
, client_name "name"
FROM
  invoice_import;

För att korrekt länka konton till kund-ID:n måste vi först känna till eller generera dessa identifierare. Låt oss lägga till fält under dem:

ALTER TABLE invoice_import ADD COLUMN client_id integer;
ALTER TABLE client_import ADD COLUMN client_id integer;

Låt oss använda metoden för att synkronisera tabeller som beskrivs ovan med ett litet tillägg - vi kommer inte att uppdatera eller ta bort något i måltabellen, eftersom vår klientimport är "enbart tillägg":

-- проставляем в таблице импорта ID уже существующих записей
UPDATE
  client_import T
SET
  client_id = D.client_id
FROM
  client D
WHERE
  T.inn = D.inn; -- unique key

-- вставляем отсутствовавшие записи и проставляем их ID
WITH ins AS (
  INSERT INTO client(
    inn
  , name
  )
  SELECT
    inn
  , name
  FROM
    client_import
  WHERE
    client_id IS NULL -- если ID не проставился
  RETURNING *
)
UPDATE
  client_import T
SET
  client_id = D.client_id
FROM
  ins D
WHERE
  T.inn = D.inn; -- unique key

-- проставляем ID клиентов у записей счетов
UPDATE
  invoice_import T
SET
  client_id = D.client_id
FROM
  client_import D
WHERE
  T.client_inn = D.inn; -- прикладной ключ

Egentligen är det allt invoice_import nu har vi anslutningsfältet ifyllt client_id, med vilken vi kommer att infoga räkningen.

Källa: will.com