DBA: organizziamo le sincronizzazioni e gli import con competenza

Nella gestione complessa di grandi quantità di dati (differenti processi ETL: importazioni, conversioni e sincronizzazioni con fonti esterne) spesso si presenta la necessità di "memorizzare" temporaneamente e elaborare rapidamente qualcosa di voluminoso.

Un compito tipico di questo tipo suona solitamente così: «Ecco qui la contabilità ha esportato dal client bancario i pagamenti ricevuti più recenti, dobbiamo caricarli velocemente sul sito e abbinarli ai conti»

Ma quando il volume di questo "qualcosa" inizia a misurarsi in centinaia di megabyte, e il servizio deve continuare a lavorare con il database in modalità 24×7, si verificano numerosi effetti collaterali che complicano la vita.

Per affrontarli in PostgreSQL (e non solo), si possono utilizzare alcune funzionalità per ottimizzazioni che permetteranno di elaborare tutto più rapidamente e con un minore consumo di risorse.

1. Dove caricare?

Innanzitutto, definiamo dove possiamo caricare i dati che vogliamo "elaborare".

1.1. Tabelle temporanee (TEMPORARY TABLE)

In linea di principio, le tabelle temporanee in PostgreSQL sono come qualsiasi altra tabella. Pertanto, sono infondate le superstizioni del tipo «lì tutto è memorizzato solo nella memoria, e questa può esaurirsi». Ma ci sono anche alcune differenze sostanziali.

Un proprio 'namespace' per ogni connessione al DB

Se due connessioni provano a eseguire simultaneamente CREATE TABLE x, allora qualcuno otterrà sicuramente un errore di non unicità degli oggetti del DB.

Ma se entrambi provano a eseguire CREARE TEMPORANEA TABELLA x, allora entrambi lo faranno correttamente, e ognuno otterrà la propria istanza della tabella. E non ci sarà nulla in comune tra loro.

«Auto-distruzione» al momento della disconnessione

Quando viene chiusa una connessione, tutte le tabelle temporanee vengono eliminate automaticamente, quindi non ha senso eseguire DROP TABLE x a meno che non sia…

Se stai lavorando tramite pgbouncer in modalità transaction, il database continua a considerare che questa connessione sia ancora attiva, e in essa questa tabella temporanea esiste ancora.

Pertanto, il tentativo di crearla nuovamente da un'altra connessione a pgbouncer porterà a un errore. Ma questo può essere aggirato, utilizzando CREA UNA TABELLA TEMPORANEA SE NON ESISTE x.

È vero, sarebbe meglio non farlo, perché si potrebbe poi "improvvisamente" scoprire di avere dati lasciati dal "precedente proprietario". È molto meglio leggere il manuale e vedere che, durante la creazione della tabella, c'è l'opzione per aggiungere ALL'IMPEGNO ELIMINA — cioè alla conclusione della transazione, la tabella verrà automaticamente eliminata.

Non-replicazione

In virtù del fatto che appartengono solo a una connessione specifica, le tabelle temporanee non vengono replicate. Tuttavia, questo elimina la necessità di una doppia scrittura dei dati in heap + WAL, quindi INSERT/UPDATE/DELETE in esse è significativamente più veloce.

Tuttavia, poiché una tabella temporanea è comunque "quasi una tabella normale", non può essere creata neanche nella replica. Almeno, per ora, anche se la relativa patch circola da tempo.

1.2. Tabelle non registrabili (UNLOGGED TABLE)

Ma cosa fare, ad esempio, se hai un qualche processo ETL ingombrante che non può essere realizzato all'interno di una singola transazione, e hai pgbouncer in modalità transaction?..

Oppure, il flusso di dati è così grande che non basta la larghezza di banda di una connessione con il DB (leggi, un processo su CPU)?..

Oppure parte delle operazioni avviene in modo asincrono in diverse connessioni?..

C'è solo un'opzione qui — creare temporaneamente una tabella non temporanea. Gioco di parole, eh. Quindi:

• ho creato "le mie" tabelle con nomi massimamente casuali per non incrociarmi con nessuno
• Estrai: ho caricato i dati da una fonte esterna
• Trasforma: ho trasformato, compilato i campi di collegamento chiave
• Load: ho trasferito i dati pronti nelle tabelle di destinazione
• ho eliminato "le mie" tabelle

E ora — un cucchiaio di catrame. In sostanza, tutta la registrazione in PostgreSQL avviene due volte — prima nel WAL, poi nei corpi delle tabelle / indici. Tutto ciò è stato fatto per supportare ACID e la corretta visibilità dei dati tra COMMIT‘nute e ROLLBACK‘nute transazioni.

Ma a noi non serve! Tutto il nostro processo è andato a buon fine nel suo insieme, oppure no. Non importa quante transazioni intermedie ci siano — non ci interessa "continuare il processo da metà", specialmente quando non è chiaro dove fosse.

A questo scopo, gli sviluppatori di PostgreSQL hanno implementato già nella versione 9.1 una funzione chiamata tabelle non registrabili (UNLOGGED):

Con questa opzione, la tabella viene creata come non registrata. I dati scritti nelle tabelle non registrate non passano attraverso il log delle scritture anticipate (vedi Capitolo 29), il che significa che queste tabelle funzionano molto più velocemente delle normali.. Tuttavia, non sono protette da guasti; in caso di guasto o spegnimento imprevisto del server, la tabella non registrata viene automaticamente tronca.Inoltre, il contenuto della tabella non registrata non viene replicato su server secondari. Gli indici creati per una tabella non registrata diventano automaticamente non registrati.

In breve, si lavorerà molto più velocemente, ma se il server di database "cade" — sarà spiacevole. Ma quanto spesso succede e il tuo processo ETL può gestirlo correttamente "dalla metà" dopo la "rinascita" del database?..

Se non è così e il caso sopra è simile al tuo — usa UNLOGGED, ma non includere mai questo attributo su tabelle reali, i cui dati ti stanno a cuore.

1.3. ON COMMIT { DELETE ROWS | DROP }

Questa costruzione consente di definire un comportamento automatico al termine della transazione quando si crea la tabella.

Riguardo ALL'IMPEGNO ELIMINA ho già scritto sopra, genera DROP TABLE, ma ecco con ALL'IMPEGNO ELIMINA RIGHE la situazione è più interessante: qui si genera TRUNCATE TABLE.

Poiché l'intera infrastruttura per la memorizzazione delle descrizioni meta delle tabelle temporanee è esattamente la stessa di quella delle normali, la continua creazione e cancellazione di tabelle temporanee porta a un forte "gonfiore" delle tabelle di sistema pg_class, pg_attribute, pg_attrdef, pg_depend,…

Ora immaginate di avere un worker connesso direttamente al DB, che ogni secondo apre una nuova transazione, crea, riempie, elabora e cancella una tabella temporanea… I rifiuti nelle tabelle di sistema si accumuleranno in eccesso, causando ritardi in ogni operazione.

In generale, non bisogna fare così! In questo caso è molto più efficace CREATE TEMPORARY TABLE x ... ON COMMIT DELETE ROWS portare fuori dal ciclo delle transazioni — allora all'inizio di ogni nuova transazione la tabella sarà già esistente (risparmiamo una chiamata CREARE), ma sarà vuota, grazie a TRUNCATE (abbiamo anche risparmiato la sua chiamata) al termine della transazione precedente.

1.4. LIKE… INCLUDING …

Ho menzionato all'inizio che uno dei casi d'uso tipici per le tabelle temporanee è rappresentato dai vari import — e lo sviluppatore copia e incolla con fatica l'elenco dei campi della tabella di destinazione nella dichiarazione della sua tabella temporanea…

Ma la pigrizia è il motore del progresso! Perciò creare una nuova tabella «per modello» può essere molto più semplice:

CREATE TEMPORARY TABLE import_table(
  LIKE target_table
);

Poiché si possono generare molteplici dati in questa tabella, le ricerche su di essa non saranno affatto veloci. Ma a questo si può porre rimedio con le tradizionali index! E, sì, anche la tabella temporanea può avere indici.

Poiché, spesso, gli indici necessari coincidono con quelli della tabella di destinazione, si può semplicemente scrivere SIMILE target_table INCLUDENDO GLI INDICI.

Se ti servono anche i DEFAULT-valori (per esempio, per riempire i valori della chiave primaria), puoi utilizzare SIMILE target_table INCLUDENDO PREDEFINITI. Oppure semplicemente — SIMILE target_table INCLUSO TUTTO — copierà i default, gli indici, i vincoli,…

Ma qui bisogna capire che se hai creato la tabella di importazione direttamente con gli indici, i dati ci metteranno più tempo a essere caricati, rispetto a caricarli prima e poi applicare gli indici — guarda come fa pg_dump.

In generale, RTFM!

2. Come scrivere?

Dico semplicemente — usa COPY-stream invece di «lotto» INSERT, velocizzazione di diversi ordini. Puoi anche farlo direttamente da un file preformattato.

3. Come elaborare?

Quindi, supponiamo che il nostro input sia più o meno così:

• Hai una tabella con i dati dei clienti su 1M di record
• ogni giorno il client ti invia un nuovo profilo completo
• dalla tua esperienza sai che, di volta in volta cambiano non più di 10K record

Un esempio classico di una tale situazione è la base KLDAR ci sono molti indirizzi, ma in ogni esportazione settimanale le modifiche (rinominazione dei comuni, unione delle strade, apparizione di nuovi edifici) sono davvero poche anche su scala nazionale.

3.1. Algoritmo di sincronizzazione completa

Per semplicità, supponiamo che non sia necessario neanche ristrutturare i dati — basta portare la tabella nel formato corretto, ovvero:

• eliminare tutto ciò che non esiste più
• aggiornare tutto ciò che c'era già e deve essere aggiornato
• inserire tutto ciò che non esisteva ancora

Perché dovresti fare le operazioni in questo ordine? Perché in questo modo la dimensione della tabella crescerà al minimo (ricorda l'MVCC!).

DELETE FROM dst

No, certo, si possono fare solo due operazioni:

• eliminare (DELETE) in realtà tutto
• inserire tutto dal nuovo profilo

Ma grazie all'MVCC, la dimensione della tabella aumenterà esattamente del doppio! Ottenere +1M di profili di record nella tabella a causa dell'aggiornamento di 10K — è un'eccesso piuttosto notevole...

TRUNCATE dst

Un sviluppatore più esperto sa che è possibile pulire l'intera tabella a un costo relativamente basso:

• pulire (TRUNCATE) l'intera tabella
• inserire tutto dal nuovo profilo

Il metodo è efficace, a volte è del tutto applicabile, ma c'è un problema… Inserire 1M di record ci vorrà un bel po', quindi non possiamo permetterci di lasciare la tabella vuota per tutto questo tempo (come accadrà senza racchiudere in un'unica transazione).

E quindi:

• comincia una transazione lunga
• TRUNCATE impone AccessExclusive-lock
• mentre facciamo l'inserimento, tutti gli altri nel frattempo non possono neppure SELECT

Non sembra andare bene…

ALTER TABLE… RENAME… / DROP TABLE …

Come alternativa, caricare tutto in una nuova tabella separata e poi semplicemente rinominare in quella vecchia. Ci sono un paio di fastidiosi dettagli:

• anche questo AccessExclusive, anche se richiede notevolmente meno tempo
• vengono annullati tutti i piani di query/statistiche di questa tabella, bisogna eseguire ANALYZE
• tutti i vincoli esterni (FK) sulla tabella

C'era una patch WIP da Simon Riggs, che proponeva di fare una operazione ALTERper sostituire il corpo della tabella a livello di file, senza toccare le statistiche e i FK, ma non raggiunse il quorum.

DELETE, UPDATE, INSERT

Quindi, ci fermiamo sulla soluzione non bloccante tra le tre operazioni. Quasi tre... Come possiamo farlo nel modo più efficace?

-- facciamo tutto nel contesto di una transazione, così nessuno vedrà stati "intermedi"
BEGIN;

-- creiamo una tabella temporanea con i dati importati
CREATE TEMPORARY TABLE tmp(
  LIKE dst INCLUDING INDEXES -- simile alla destinazione, inclusi gli indici
) ON COMMIT DROP; -- non ci serve fuori dalla transazione

-- inseriamo rapidamente il nuovo set di dati tramite COPY
COPY tmp FROM STDIN;
-- ...
-- .

-- rimuoviamo i mancanti
DELETE FROM
  dst D
USING
  dst X
LEFT JOIN
  tmp Y
    USING(pk1, pk2) -- campi della chiave primaria
WHERE
  (D.pk1, D.pk2) = (X.pk1, X.pk2) AND
  Y IS NOT DISTINCT FROM NULL; -- "anti-join"

-- aggiorniamo i rimanenti
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); -- non c'è bisogno di aggiornare le corrispondenze

-- inseriamo i mancanti
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. Post-elaborazione dell'importazione

Nello stesso KADRE, tutte le registrazioni modificate devono essere ulteriormente elaborate — normalizzate, estratte le parole chiave, convertite nelle strutture necessarie. Ma come si fa a sapere — cosa sia stato esattamente modificato, senza complicare il codice di sincronizzazione, idealmente, non toccandolo affatto?

Se il processo di scrittura durante la sincronizzazione è esclusivo per il vostro processo, si può utilizzare un trigger che raccoglierà tutte le modifiche per noi:

-- tabelle di destinazione
CREATE TABLE kladr(...);
CREATE TABLE kladr_house(...);

-- tabelle con la cronologia delle modifiche
CREATE TABLE kladr$log(
  ro kladr, -- qui si trovano le immagini intere dei record vecchi/nuovi
  rn kladr
);

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

-- funzione generale di registrazione delle modifiche
CREATE OR REPLACE FUNCTION diff$log() RETURNS trigger AS $$
DECLARE
  dst varchar = TG_TABLE_NAME || '$log';
  stmt text = '';
BEGIN
  -- controlliamo la necessità di registrare durante l'aggiornamento di un record
  IF TG_OP = 'UPDATE' THEN
    IF NEW IS NOT DISTINCT FROM OLD THEN
      RETURN NEW;
    END IF;
  END IF;
  -- creiamo un record di log
  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;

Ora possiamo applicare i trigger prima dell'inizio della sincronizzazione (o abilitarli tramite 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();

E poi possiamo estrarre tranquillamente tutte le modifiche necessarie dalle tabelle di log e passarle attraverso ulteriori elaboratori.

3.3. Importazione di insiemi di dati correlati

In precedenza abbiamo esaminato i casi in cui le strutture dati della sorgente e del destinatario sono coincidenti. Ma cosa fare se l'esportazione da un sistema esterno ha un formato diverso dalla struttura di archiviazione nel nostro database?

Prendiamo come esempio la conservazione dei clienti e delle relative fatture, classico caso di "molti-a-uno":

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

Ma ecco che l'esportazione da una fonte esterna arriva a noi in forma di "tutto in uno":

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

È evidente che i dati sui clienti possono essere duplicati in questo caso, e la registrazione principale è "fattura":

0123456789;Vasya;A-01;2020-03-16;1000.00
9876543210;Petya;A-02;2020-03-16;666.00
0123456789;Vasya;B-03;2020-03-16;9999.00

Per il modello, inseriamo semplicemente i nostri dati di test, ma ricordiamo — COPY è più efficiente!

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

Prima di tutto, identifichiamo i 'tagli' a cui i nostri 'fatti' si riferiscono. Nel nostro caso, le fatture si riferiscono ai clienti:

CREATE TEMPORARY TABLE client_import AS
SELECT DISTINCT ON(client_inn)
-- è possibile usare semplicemente SELECT DISTINCT se i dati sono già coerenti
  client_inn inn
, client_name "name"
FROM
  invoice_import;

Per collegare correttamente le fatture agli ID dei clienti, dobbiamo prima ottenere o generare questi identificatori. Aggiungiamo i campi per essi:

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

Utilizzeremo il metodo di sincronizzazione descritto sopra con una leggera modifica: non aggiorneremo né cancelleremo nulla nella tabella di destinazione, poiché l'importazione dei clienti è 'append-only':

-- impostiamo nella tabella di importazione l'ID delle registrazioni esistenti
UPDATE
  client_import T
SET
  client_id = D.client_id
FROM
  client D
WHERE
  T.inn = D.inn; -- chiave unica

-- inseriamo le registrazioni mancanti e impostiamo i loro ID
WITH ins AS (
  INSERT INTO client(
    inn
  , name
  )
  SELECT
    inn
  , name
  FROM
    client_import
  WHERE
    client_id IS NULL -- se l'ID non è stato impostato
  RETURNING *
)
UPDATE
  client_import T
SET
  client_id = D.client_id
FROM
  ins D
WHERE
  T.inn = D.inn; -- chiave unica

-- impostiamo l'ID dei clienti per le registrazioni delle fatture
UPDATE
  invoice_import T
SET
  client_id = D.client_id
FROM
  client_import D
WHERE
  T.client_inn = D.inn; -- chiave applicativa

In sostanza, tutto — in invoice_import ora abbiamo popolato il campo di collegamento client_id, con cui inseriremo la fattura.

Fonte: habr.com