DBA: Synchronisationen und Importe sinnvoll organisieren

Bei der komplexen Verarbeitung großer Datensätze (verschiedene ETL-Prozesse: Importe, Konvertierungen und Synchronisationen mit externen Quellen) entsteht oft die Notwendigkeit temporär "zu speichern" und sofort schnell zu verarbeiten etwas Umfangreiches.

Eine typische Aufgabe dieser Art lautet im Allgemeinen etwa so: "Hier hat die Buchhaltung die letzten eingegangenen Zahlungen aus dem Kundenbankupdate exportiert, die müssen schnell auf die Website hochgeladen und den Konten zugewiesen werden" Doch wenn das Volumen dieses "etwas" in Hunderten von Megabytes gemessen wird und der Dienst dabei 24 × 7 mit der Datenbank arbeiten muss, treten viele Nebenwirkungen auf, die Ihr Leben erschweren.

Um mit ihnen in PostgreSQL (und nicht nur darin) umzugehen, kann man einige Optimierungsfunktionen nutzen, die eine schnellere Verarbeitung bei geringerem Ressourcenaufwand ermöglichen.

1. Wo hochladen?

Zunächst sollten wir klären, wohin wir die Daten hochladen können, die wir "verarbeiten" wollen.

1.1. Temporäre Tabellen (TEMPORARY TABLE)

Prinzipiell sind temporäre Tabellen in PostgreSQL ganz normale Tabellen wie alle anderen auch. Daher sind Aberglauben wie

"dort wird alles nur im Speicher gespeichert, und der kann ausgehen" nicht zutreffend. Aber es gibt auch einige wesentliche Unterschiede.Eine eigene "Namespaces" für jede Verbindung zur Datenbank

Wenn zwei Verbindungen gleichzeitig versuchen,

CREATE TABLE x , dann wird einer von ihnen definitiveinen Fehler der Nicht-Eindeutigkeit von DB-Objekten erhalten. Wenn jedoch beide versuchen,

, wird jeder dies ordnungsgemäß durchführen und jeder wird CREATE VORÜBERGEHEND TABELLE xseinen eigenen Instanz der Tabelle erhalten. Und es wird nichts Gemeinsames zwischen ihnen geben. "Selbstzerstörung" bei der Trennung

Beim Schließen der Verbindung werden alle temporären Tabellen automatisch gelöscht, daher macht es "von Hand"

DROP TABLE x keinen Sinn, außer ... Wenn Sie über

pgbouncer im Transaktionsmodus , dann denkt die Datenbank weiterhin, dass diese Verbindung noch aktiv ist, und in ihr existiert diese temporäre Tabelle nach wie vor.Deshalb wird der Versuch, sie erneut aus einer anderen Verbindung zu pgbouncer zu erstellen, zu einem Fehler führen. Aber das lässt sich umgehen, indem man

Es ist jedoch besser, so etwas zu vermeiden, da man dann "unerwartet" Daten finden könnte, die vom "vorherigen Besitzer" übrig geblieben sind. Stattdessen ist es viel besser, tatsächlich das Handbuch zu lesen und zu sehen, dass man bei der Erstellung einer Tabelle die Möglichkeit hat, diese zu ergänzen. TEMPORÄRE TABELLE ERSTELLEN WENN NICHT EXISTIERT x.

Natürlich ist es besser, dies so nicht zu machen, da man dann „plötzlich“ Daten findet, die vom „vorherigen Besitzer“ übrig geblieben sind. Stattdessen ist es viel besser, das Handbuch zu lesen und zu sehen, dass beim Erstellen einer Tabelle die Möglichkeit besteht, ergänzende Informationen hinzuzufügen. BEI BESTÄTIGUNG DROP — das heißt, nach Abschluss der Transaktion wird die Tabelle automatisch gelöscht.

Nicht-Replikation

Aufgrund der Zugehörigkeit zu einer bestimmten Verbindung werden temporäre Tabellen nicht repliziert. Dafür befreit es von der Notwendigkeit der doppelten Datenspeicherung in Heap + WAL, weshalb INSERT/UPDATE/DELETE darin erheblich schneller ist.

Aber da eine temporäre Tabelle dennoch eine "fast gewöhnliche" Tabelle ist, kann sie auch auf der Replik nicht erstellt werden. Zumindest im Moment, obwohl der entsprechende Patch schon lange existiert.

1.2. Nicht-protokollierte Tabellen (UNLOGGED TABLE)

Was ist jedoch zu tun, wenn Sie beispielsweise einen ziemlich schwerfälligen ETL-Prozess haben, den Sie nicht innerhalb einer einzigen Transaktion realisieren können, und Sie dennoch , dann denkt die Datenbank weiterhin, dass diese Verbindung noch aktiv ist, und in ihr existiert diese temporäre Tabelle nach wie vor.?..

Oder wenn der Datenstrom so groß ist, dass die Bandbreite einer einzigen Verbindung zur DB (sprich, eines Prozesses auf der CPU) nicht ausreicht? ..

Oder wenn Teile der Operationen asynchron in verschiedenen Verbindungen durchgeführt werden? ..

Hier gibt es nur eine Möglichkeit — vorübergehend eine Nicht-Temporäre Tabelle zu erstellen. Wortspiel, ja. Das heißt:

• Ich habe "meine" Tabellen mit maximal zufälligen Namen erstellt, um nicht mit jemandem in Konflikt zu geraten
• Extraktion: Ich habe Daten aus einer externen Quelle in sie geladen
• Transformation: Ich habe sie umgewandelt und die Schlüsselfelder gefüllt
• mit dem ausgewählten Testplan: Ich habe die fertigen Daten in die Zieltabellen übertragen
• und meine Tabellen gelöscht

Und jetzt — ein Wermutstropfen. Im Wesentlichen werden alle Schreibvorgänge in PostgreSQL zweimal durchgeführt — zuerst im WAL, dann in die Tabellen- / Indexkörper. All dies wurde zur Unterstützung von ACID und zur korrekten Sichtbarkeit der Daten zwischen COMMITden ‘geöffneten und ROLLBACK‘geöffneten Transaktionen.

Aber das brauchen wir nicht! Bei uns ist der gesamte Prozess entweder vollständig erfolgreich oder nicht. Es spielt keine Rolle, wie viele Zwischentransaktionen darin stattfinden — uns interessiert nicht, „den Prozess von der Mitte aus fortzusetzen“, besonders wenn unklar ist, wo sie war.

Dafür haben die Entwickler von PostgreSQL bereits in Version 9.1 eine Funktion eingeführt, wie nicht-protokollierte (UNLOGGED) Tabellen:

Mit diesem Hinweis wird die Tabelle als nicht-protokolliert erstellt. Die Daten, die in nicht-protokollierte Tabellen geschrieben werden, durchlaufen nicht das Protokoll der Voraufzeichnung (siehe Kapitel 29), wodurch solche Tabellen deutlich schneller arbeiten als gewöhnliche.Sie sind jedoch nicht vor einem Ausfall geschützt; bei einem Ausfall oder einem plötzlichen Serverabbruch wird die nicht-protokollierte Tabelle automatisch abgeschnitten.Darüber hinaus wird der Inhalt der nicht-protokollierten Tabelle nicht repliziert. auf die verwalteten Server. Alle Indizes, die für eine nicht protokollierte Tabelle erstellt werden, werden automatisch nicht protokolliert.

Kurz gesagt, wird es deutlich schneller sein,, aber wenn der DB-Server „ausfällt“, wird es unangenehm. Kommt das häufig vor, und ist Ihr ETL-Prozess in der Lage, dies korrekt „aus der Mitte“ nach „Belebung“ der DB zu verarbeiten?..

Wenn nicht, und der oben genannte Fall Ihrem ähnelt - verwenden Sie UNLOGGED, aber aktivieren Sie dieses Attribut niemals für reale Tabellen , deren Daten Ihnen wichtig sind.1.3. ON COMMIT { DELETE ROWS | DROP }

Diese Konstruktion ermöglicht es, beim Erstellen einer Tabelle ein automatisches Verhalten beim Abschluss der Transaktion festzulegen.

Ich habe bereits oben geschrieben, dass er generiert,

Über BEI BESTÄTIGUNG DROP , aber die Situation mit DROP TABLEinteressanter - hier wird erzeugt BEI BESTÄTIGUNG ZEILE LÖSCHEN TRUNCATE TABLE Da die gesamte Infrastruktur zur Speicherung der Metadaten der temporären Tabelle genau die gleiche ist wie die einer normalen Tabelle, führt.

das ständige Erstellen und Löschen temporärer Tabellen zu einem starken "Aufblähen" der Systemtabellen pg_class, pg_attribute, pg_attrdef, pg_depend,… Stellen Sie sich jetzt vor, Sie haben einen Worker, der über eine direkte Verbindung zur DB jede Sekunde eine neue Transaktion öffnet, eine temporäre Tabelle erstellt, füllt, verarbeitet und löscht… Der Müll in den Systemtabellen wird sich gehäuft ansammeln, und das führt zu zusätzlichen Verzögerungen bei jeder Operation.

Im Allgemeinen sollte man das vermeiden! In diesem Fall ist es viel effizienter,

CREATE TEMPORARY TABLE x ... ON COMMIT DELETE ROWS außerhalb des Transaktionszyklus zu halten - dann wird die Tabelle zu Beginn jeder neuen Transaktion bereits existieren (wir sparen den Aufruf ), aber CREATEwird leer sein , dank(auch diesen Aufruf haben wir gespart) beim Abschluss der vorherigen Transaktion. TRUNCATE 1.4. LIKE… INCLUDING …

Ich habe zu Beginn erwähnt, dass einer der typischen Anwendungsfälle für temporäre Tabellen verschiedene Arten von Importen sind - und der Entwickler kopiert und fügt müde die Liste der Felder der Zieltabelle in die Deklaration seiner temporären Tabelle ein...

Aber Faulheit ist der Motor des Fortschritts! Daher

kann eine neue Tabelle „nach Vorbild“ viel einfacher erstellt werden: CREATE TEMPORARY TABLE import_table( LIKE target_table );

Da in diese Tabelle später ziemlich viele Daten eingefügt werden können, wird die Suche darin alles andere als schnell sein. Aber dagegen gibt es eine traditionelle Lösung - Indizes! Und ja,

auch temporäre Tabellen können Indizes haben. Da oft die benötigten Indizes mit den Indizes der Zieltabelle übereinstimmen, kann man einfach schreiben,.

Wenn Sie auch LIKE Ziel_Tabelle EINSCHLIESSLICH INDEXEN.

DEFAULT STANDARD-Werte (zum Beispiel zur Befüllung der Werte des Primärschlüssels) kann man nutzen LIKE Ziel_Tabelle EINSCHLIESSLICH DER STANDARDWERT. Oder einfach — LIKE Ziel_Tabelle EINSCHLIESSLICH ALLER — kopiert die Standardwerte, Indizes, Constraints,…

Aber hier muss man schon verstehen, dass wenn Sie die Importtabelle sofort mit Indizes erstellt haben, die Daten länger benötigt werden, als wenn man zuerst alles importiert und dann die Indizes hinzufügt — siehe als Beispiel, wie das pg_dump.

Im Allgemeinen, RTFM!

2. Wie schreiben?

Ich sage einfach — verwenden Sie COPY-Stream statt "Batch" INSERT, Beschleunigung um ein Vielfaches. Man kann sogar direkt aus einer vorher formatierten Datei.

3. Wie verarbeiten?

Also, nehmen wir an, dass unser Input ungefähr so aussieht:

• Sie haben in Ihrer Datenbank eine Tabelle mit Kundendaten von 1M Datensätzen
• jeden Tag sendet der Kunde Ihnen ein neues volles "Abbild"
• Aus Erfahrung wissen Sie, dass sich von Mal zu Mal nicht mehr als 10K Datensätze ändern

Ein klassisches Beispiel für eine solche Situation ist die KLRAD-Datenbank — es gibt viele Adressen, aber in jedem wöchentlichen Update an Änderungen (Umbenennungen von Städten, Fusionen von Straßen, das Auftauchen neuer Häuser) gibt es insgesamt sehr wenige, selbst im Maßstab des gesamten Landes.

3.1. Algorithmus zur vollständigen Synchronisierung

Zur Vereinfachung nehmen wir an, dass es nicht nötig ist, die Daten neu zu strukturieren — bringen Sie einfach die Tabelle in die gewünschte Form, das heißt:

• löschen alles, was nicht mehr vorhanden ist
• das System aus dem p9-Zweig des Repositories „Sisyphus“ aktualisieren. Für neue Unternehmenskunden besteht die Möglichkeit, Testversionen zu erhalten, und Privatnutzern wird traditionell angeboten, die benötigte Version des OS „Alt“ kostenlos von der Website „Basalt SPO“ oder von der neuen Download-Seite alles, was bereits war und aktualisiert werden muss
• einfügen alles, was noch nicht vorhanden war

Warum sollte man die Operationen genau in dieser Reihenfolge durchführen? Weil die Tabellengröße auf diese Weise minimal wachsen wird (denken Sie an MVCC!).

DELETE FROM dst

Nein, natürlich kann man mit nur zwei Operationen auskommen:

• löschen (DELETE) tatsächlich alles
• einfügen alles aus dem neuen Abbild

Aber dabei wird, dank MVCC, die Tabellengröße genau um das Zweifache wachsen! +1M Abbilden von Datensätzen in der Tabelle aufgrund von 10K-Updates ist ziemlich übertrieben…

TRUNCATE dst

Ein erfahrener Entwickler weiß, dass man die gesamte Tabelle relativ kostengünstig leeren kann:

• bereinigen (TRUNCATE) die gesamte Tabelle
• einfügen alles aus dem neuen Abbild

Die Methode ist effektiv, in manchen Fällen durchaus anwendbar, aber es gibt ein Problem… 1M Datensätze wird viel Zeit in Anspruch nehmen, sodass wir es uns nicht erlauben können, die Tabelle für all die Zeit leer zu lassen (wie es ohne Umschließung in eine einzige Transaktion geschehen würde).

Das bedeutet:

• wir beginnen eine lange Transaktion
• TRUNCATE legt AccessExclusive- die Sperrung
• wir machen lange Einfügungen, während alle anderen in der Zwischenzeit nicht einmal SELECT

Es sieht nicht gut aus…

ALTER TABLE… UMBENENNEN… / TABELLE LÖSCHEN …

Eine Möglichkeit wäre, alles in eine separate neue Tabelle zu laden und dann einfach auf den Platz der alten zu benennen. Ein paar unangenehme Kleinigkeiten:

• doch auch AccessExclusive, wenn auch erheblich weniger zeitaufwendig
• alle Abfragepläne / Statistiken dieser Tabelle werden zurückgesetzt, muss ANALYZE durchgeführt werden
• alle Fremdschlüssel (FK) auf die Tabelle

Es gab einen WIP-Patch von Simon Riggs, der vorschlug, eine ALTER-Operation für den Austausch des Tabellenkörpers auf Dateiebene vorzunehmen, ohne die Statistiken und FK zu berühren, aber er konnte kein quorum erreichen.

DELETE, UPDATE, INSERT

Also entscheiden wir uns für die nicht-blockierende Variante aus drei Operationen. Fast drei… Wie lässt sich das am effizientesten umsetzen?

-- alles im Rahmen einer Transaktion durchführen, damit niemand die "zwischenzeitlichen" Zustände sieht
BEGIN;

-- erstellen einer temporären Tabelle mit importierten Daten
CREATE TEMPORARY TABLE tmp(
  LIKE dst INCLUDING INDEXES -- nach dem Vorbild, einschließlich Indizes
) ON COMMIT DROP; -- außerhalb der Transaktion benötigen wir sie nicht

-- schnell neuen Datensatz per COPY einfügen
COPY tmp FROM STDIN;
-- ...
-- .

-- fehlende Datensätze entfernen
DELETE FROM
  dst D
USING
  dst X
LEFT JOIN
  tmp Y
    USING(pk1, pk2) -- Felder des Primärschlüssels
WHERE
  (D.pk1, D.pk2) = (X.pk1, X.pk2) AND
  Y IS NOT DISTINCT FROM NULL; -- "Anti-Join"

-- verbleibende aktualisieren
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); -- unnötig Übereinstimmungen zu aktualisieren

-- fehlende einfügen
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. Nachverarbeitung des Imports

Im gleichen KLRAD müssen alle geänderten Einträge zusätzlich durch die Nachbearbeitung laufen — normalisiert, Schlüsselwörter extrahiert, in die erforderlichen Strukturen gebracht. Aber wie kann man herausfinden — was genau geändert wurde, ohne dabei den Synchronisierungscode zu verkomplizieren, idealerweise überhaupt nicht zu berühren?

Wenn nur Ihr Prozess zur Synchronisierungszeit Schreibzugriff hat, können Sie einen Trigger verwenden, der alle Änderungen für uns erfasst:

-- Zieltabelle
CREATE TABLE kladr(...);
CREATE TABLE kladr_house(...);

-- Tabellen mit Änderungsprotokoll
CREATE TABLE kladr$log(
  ro kladr, -- hier liegen vollständige Abbilder der alten/neuen Einträge
  rn kladr
);

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

-- allgemeine Funktion zur Protokollierung von Änderungen
CREATE OR REPLACE FUNCTION diff$log() RETURNS trigger AS $$
DECLARE
  dst varchar = TG_TABLE_NAME || '$log';
  stmt text = '';
BEGIN
  -- Überprüfen, ob Protokollierung bei der Aktualisierung des Eintrags erforderlich ist
  IF TG_OP = 'UPDATE' THEN
    IF NEW IS NOT DISTINCT FROM OLD THEN
      RETURN NEW;
    END IF;
  END IF;
  -- Eintrag im Protokoll erstellen
  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;

Jetzt können wir die Trigger vor Beginn der Synchronisierung anwenden (oder aktivieren über 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();

Und dann extrahieren wir einfach aus den log-Tabellen alle erforderlichen Änderungen und schicken sie an zusätzliche Bearbeitungsmechanismen.

3.3. Import von verknüpften Datensätzen

Wir haben oben Fälle betrachtet, in denen die Datenstrukturen von Quelle und Empfänger übereinstimmen. Aber was tun, wenn der Export aus dem externen System ein Format hat, das von der Speicherung in unserer Datenbank abweicht?

Nehmen wir als Beispiel die Speicherung von Kunden und deren Rechnungen, den klassischen Fall «viele-zu-eins»:

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

Und die Daten von der externen Quelle kommen bei uns im Format 'alles in einem':

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

Offensichtlich können die Kundendaten in diesem Format dupliziert werden, wobei die Hauptzeile 'Rechnung' ist:

0123456789;Wasi;A-01;2020-03-16;1000.00
9876543210;Petja;A-02;2020-03-16;666.00
0123456789;Wasi;B-03;2020-03-16;9999.00

Für das Modell fügen wir einfach unsere Testdaten ein, aber denken wir daran — COPY effizienter!

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

Zuerst identifizieren wir die 'Schnitte', auf die sich unsere 'Fakten' beziehen. In unserem Fall beziehen sich Rechnungen auf Kunden:

ERSTELLE TEMPORÄRE TABELLE client_import WIE
WÄHLE Eindeutig AUF(client_inn)
-- man kann einfach SELECT DISTINCT verwenden, wenn die Daten bereits konsistent sind
  client_inn inn
, client_name "name"
FROM
  invoice_import;

Um die Rechnungen korrekt mit den Kunden-IDs zu verknüpfen, müssen wir diese Identifikatoren zunächst herausfinden oder generieren. Lassen Sie uns dafür Felder hinzufügen:

ALTER TABLE invoice_import HINZUFÜGEN SPALTE client_id integer;
ALTER TABLE client_import HINZUFÜGEN SPALTE client_id integer;

Wir werden die oben beschriebene Methode zur Synchronisierung der Tabellen mit einer kleinen Anpassung verwenden – wir werden nichts in der Zieltabelle aktualisieren oder löschen, da der Import von Kunden bei uns „append-only“ ist:

-- wir setzen in der Importtabelle die IDs der bereits vorhandenen Datensätze
UPDATE
  client_import T
SET
  client_id = D.client_id
FROM
  client D
WHERE
  T.inn = D.inn; -- eindeutiger Schlüssel

-- fügen Sie die fehlenden Datensätze ein und setzen Sie deren IDs
MIT ins ALS (
  INSERT INTO client(
    inn
  , name
  )
  WÄHLE
    inn
  , name
  FROM
    client_import
  WHERE
    client_id IS NULL -- wenn die ID nicht gesetzt wurde
  RETURNING *
)
UPDATE
  client_import T
SET
  client_id = D.client_id
FROM
  ins D
WHERE
  T.inn = D.inn; -- eindeutiger Schlüssel

-- wir setzen die Kunden-IDs bei den Rechnungsdatensätzen
UPDATE
  invoice_import T
SET
  client_id = D.client_id
FROM
  client_import D
WHERE
  T.client_inn = D.inn; -- anwendungsbezogener Schlüssel

Im Grunde ist alles – in invoice_import ist jetzt das Verknüpfungsfeld ausgefüllt client_id, mit dem wir die Rechnung einfügen werden.

Quelle: habr.com