DBA: wir organisieren Synchronisierungen und Importe geschickt

Bei der komplexen Verarbeitung großer Datensätze (verschiedene ETL-Prozesse: Importe, Konversionen und Synchronisationen mit externen Quellen) besteht häufig die Notwendigkeit, etwas vorübergehend zu "merken" und sofort schnell zu verarbeiten, was auch immer umfangreich ist.

Typische Aufgaben in diesem Bereich lauten normalerweise etwa so: "Hier hat die Buchhaltung die letzten Zahlungen aus dem Kundenbank abgeholt, wir müssen diese schnell auf die Website laden und mit den Konten verknüpfen." Wenn das Volumen dieses "Irgendwas" jedoch in Hunderten von Megabyte gemessen wird und der Dienst dabei 24/7 mit der Datenbank arbeiten muss, entstehen viele Nebeneffekte, die Ihnen das Leben schwer machen.

Um damit in PostgreSQL (und auch nicht nur dort) umzugehen, können einige Optimierungsmöglichkeiten genutzt werden, die eine schnellere Verarbeitung mit geringerem Ressourcenverbrauch ermöglichen.

1. Wohin laden?

Lassen Sie uns zunächst klären, wohin wir die Daten hochladen können, die wir "verarbeiten" möchten.

1.1. Temporäre Tabellen (TEMPORARY TABLE)

Im Grunde genommen sind temporäre Tabellen in PostgreSQL genau wie jede andere Tabelle. Aberglaube wie

Im Prinzip sind temporäre Tabellen in PostgreSQL genau wie alle anderen Tabellen. Daher sind Aberglauben wie Hier wird alles nur im Speicher gespeichert, und der kann schnell voll sein.Es gibt jedoch auch einige wesentliche Unterschiede.

Ein eigener "Namespace" für jede Datenbankverbindung.

Wenn zwei Verbindungen gleichzeitig versuchen, CREATE TABLE x, wird einer von beiden garantiert einen Fehler wegen Nicht-Eindeutigkeit der Datenbankobjekte erhalten.

Wenn jedoch beide versuchen, ERSTELLEN TEMPORÄR TABELLE x, dann können beide das erfolgreich durchführen, und jede Verbindung erhält ihren eigenen Instanz der Tabelle. Und zwischen ihnen wird es keine Gemeinsamkeiten geben.

„Selbstzerstörung“ bei der Trennung

Wenn die Verbindung geschlossen wird, werden alle temporären Tabellen automatisch gelöscht, daher macht es keinen Sinn, DROP TABLE x manuell auszuführen, außer...

Wenn Sie über pgbouncer im Transaktionsmodus arbeiten,denn die Datenbank denkt weiterhin, dass diese Verbindung noch aktiv ist, und in dieser existiert diese temporäre Tabelle nach wie vor.

Daher 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 TEMPORÄRE TABELLE ERSTELLEN WENN NICHT EXISTIERT x.

Es ist jedoch besser, dies zu vermeiden, da man sonst möglicherweise plötzlich Daten vom "vorherigen Besitzer" entdecken könnte. Stattdessen ist es weitaus besser, das Handbuch zu lesen und zu erkennen, dass es beim Erstellen einer Tabelle die Möglichkeit gibt, diese zu ergänzen. BEIMEN WIDMEN LÖSCHEN — das heißt, dass die Tabelle nach Abschluss der Transaktion automatisch gelöscht wird.

Keine Replikation

Aufgrund der Zugehörigkeit zu einer bestimmten Verbindung werden temporäre Tabellen nicht repliziert. Allerdings entfernt dies die Notwendigkeit einer doppelten Datenspeicherung in Heap + WAL, weshalb INSERT/UPDATE/DELETE deutlich schneller ist.

Aber da eine temporäre Tabelle ja dennoch "fast wie eine normale" Tabelle ist, kann sie auch nicht auf der Replik erstellt werden. Zumindest bislang nicht, obwohl der entsprechende Patch schon lange bekannt ist.

1.2. Nicht protokollierte Tabellen (UNLOGGED TABLE)

Aber was tun, wenn Sie beispielsweise einen umfangreichen ETL-Prozess haben, der nicht innerhalb einer einzigen Transaktion realisiert werden kann, und Sie dennoch pgbouncer im Transaktionsmodus arbeiten,?..

Oder wenn der Datenstrom so groß ist, dass die Bandbreite einer Verbindung mit der Datenbank nicht ausreicht (lesen Sie: ein Prozess pro CPU)?

Oder wenn Teile der Operationen asynchron in verschiedenen Verbindungen ablaufen?

In diesem Fall gibt es nur eine Möglichkeit — temporäre nicht-temporäre Tabellen erstellen. Wortspiel, oder? Das bedeutet:

• hat seine eigenen Tabellen mit maximal zufälligen Namen erstellt, um keine Überschneidungen zu haben
• Extrahieren: habe Daten aus einer externen Quelle hochgeladen
• Transformieren: habe die Schlüsselverknüpfungsfelder ausgefüllt
• Load: habe die fertigen Daten in die Zieltabellen übertragen
• habe meine eigenen Tabellen gelöscht

Und jetzt — ein Wermutstropfen. Im Grunde genommen findet jede Aufzeichnung in PostgreSQL zweimal statt — zuerst im WAL, dann in den Tabellen-/Indexkörpern. All dies geschieht zur Unterstützung von ACID und einer korrekten Sichtbarkeit der Daten zwischen BESTÄTIGENverschachtelten und ROLLBACKverschachtelten Transaktionen.

Aber das brauchen wir nicht! Der gesamte Prozess ist entweder erfolgreich abgeschlossen oder nicht.Es spielt keine Rolle, wie viele Zwischentransaktionen es gibt — uns interessiert nicht, „den Prozess von der Mitte aus fortzusetzen“, insbesondere wenn unklar ist, wo er war.

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

ermöglicht. Mit dieser Angabe wird die Tabelle als nicht protokolliert erstellt. Die Daten, die in nicht protokollierte Tabellen geschrieben werden, durchlaufen nicht das Write-Ahead-Log (siehe Kapitel 29), was diese Tabellen arbeiten wesentlich schneller als gewöhnliche.. Sie sind jedoch nicht vor Ausfällen geschützt; bei einem Ausfall oder einer Notabschaltung des Servers wird die unlogische Tabelle automatisch gekappt.. Außerdem wird der Inhalt der unlogischen Tabelle nicht auf die nachfolgenden Server repliziert. Jegliche Indizes, die für die unlogische Tabelle erstellt werden, werden automatisch unlogisch. Kurz gesagt,

wird es erheblich schneller sein , aber wenn der DB-Server 'ausfällt' – ist das unangenehm. Doch wie oft passiert das, und kann Ihr ETL-Prozess dies korrekt 'von der Mitte' nach dem 'Wachruf' der DB handhaben?..Wenn das nicht der Fall ist und der oben genannte Fall Ihrem ähnelt – verwenden Sie

UNLOGGED , aber aktivieren Sie niemalsdieses Attribut bei realen Tabellen , deren Daten Ihnen wichtig sind.1.3. ON COMMIT { DELETE ROWS | DROP }

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

Ich habe bereits oben geschrieben, dass er generiert

Über BEIMEN WIDMEN LÖSCHEN , während es mit DROP TABLEinteressanter aussieht – hier wird generiert BEIMEN WIDMEN ZEILE LÖSCHEN TRUNCATE TABLE Da die gesamte Infrastruktur zur Speicherung der Metadaten der temporären Tabelle genau die gleiche wie die der normalen ist, ist.

Da die gesamte Infrastruktur zur Speicherung der Metadaten temporärer Tabellen genau so ist wie bei regulären Tabellen, ist Das ständige Erstellen und Löschen von temporären Tabellen führt zu einer erheblichen "Aufblähung" der Systemtabellen. pg_class, pg_attribute, pg_attrdef, pg_depend,…

Stellen Sie sich vor, Sie haben einen Worker mit einer direkten Verbindung zur Datenbank, der 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 anhäufen und es gibt unnötige Verzögerungen bei jeder Operation.

Im Grunde genommen, so sollte man das nicht machen! In diesem Fall ist es viel effizienter CREATE TEMPORARY TABLE x ... ON COMMIT DELETE ROWS außerhalb des Transaktionszyklus zu erstellen — dann existieren die Tabellen bereits zu Beginn jeder neuen Transaktion. Sie wird existieren (wir sparen den Aufruf ERSTELLEN), aber sie wird leer sein, dank TRUNCATE (diesen Aufruf haben wir ebenfalls gespart) am Ende der vorherigen Transaktion.

1.4. LIKE… INCLUDING …

Ich habe zu Beginn erwähnt, dass ein typischer Anwendungsfall für temporäre Tabellen verschiedene Arten von Importen sind — und der Entwickler kopiert müde die Liste der Felder der Zieltabelle in die Deklaration seiner temporären Tabelle…

Aber Faulheit ist der Motor des Fortschritts! Daher kann eine neue Tabelle "nach Muster" viel einfacher erstellt werden:

CREATE TEMPORARY TABLE import_table(
  LIKE target_table
);

Da es möglich ist, in diese Tabelle eine große Menge an Daten zu generieren, 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: WIE target_table EINSCHLIESSLICH INDEXE.

Wenn Sie auch DEFAULT-Werte (zum Beispiel zur Auffüllung der Werte des Primärschlüssels) benötigen, können Sie WIE target_table EINSCHLIEßLICH VORGABENnutzen. Oder einfach – WIE target_table EINSCHLIEßLICH ALLER — kopiert die Defaults, Indizes, Constraints,…

Aber hier muss man schon verstehen, dass, wenn Sie die Importtabelle gleich mit Indizes erstellt haben, die Datenübertragung länger dauern wird , als wenn Sie zuerst alles importieren und dann die Indizes anwenden – schauen Sie sich als Beispiel an, wie das gemacht wird vonIm Allgemeinen, pg_dump.

RTFM 2. Wie schreibt man?!

Ich sage es einfach – benutzen Sie

-Stream anstelle von "Batch" COPYdas beschleunigt um ein Vielfaches. INSERT, Man kann sogar direkt aus einer vorher formatierten Datei.3. Wie verarbeitet man?

Lassen Sie uns also annehmen, unser Einstieg sieht etwa so aus:

Sie haben in Ihrer Datenbank eine Tabelle mit Kundendaten mit

• 1 Million Datensätzen. Jeden Tag sendet Ihnen der Kunde ein neues
• vollständiges "Abbild". Aus Erfahrung wissen Sie, dass es von Mal zu Mal
• aus Erfahrung wissen Sie, dass es von Mal zu Mal Es werden nicht mehr als 10.000 Datensätze geändert

Ein klassisches Beispiel für eine solche Situation ist die KLGDR-Datenbank — es gibt viele Adressen, aber in jeder wöchentlichen Exportdatei gibt es nur sehr wenige Änderungen (Umbenennungen von Orten, Zusammenlegungen von Straßen, das Erscheinen neuer Gebäude), selbst im Maßstab des ganzen Landes.

3.1. Algorithmus für die vollständige Synchronisation

Um es einfach zu halten, nehmen wir an, dass Sie die Daten nicht einmal umstrukturieren müssen — bringen Sie die Tabelle einfach in die gewünschte Form, das heißt:

• löschen alles, was nicht mehr existiert
• aktualisieren alles, was bereits vorhanden war und aktualisiert werden muss
• einfügen alles, was noch nicht vorhanden war

Warum sollten diese Operationen genau in dieser Reihenfolge durchgeführt werden? Weil auf diese Weise die Größe der Tabelle minimal wachsen wird (denke an MVCC!).

DELETE FROM dst

Natürlich kann man auch mit nur zwei Operationen auskommen:

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

Aber dabei wird dank MVCC die Größe der Tabelle genau doppelt so groß! +1 Million Datensätze in der Tabelle durch das Aktualisieren von 10.000 — das ist eine bescheidene Überflüssigkeit…

TRUNCATE dst

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

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

Die Methode ist effektiv, manchmal durchaus anwendbar, aber es gibt ein Problem… Das Einfüllen von 1 Million Datensätzen wird lange dauern, daher können wir es uns nicht leisten, die Tabelle während dieser Zeit leer zu lassen (wie es ohne eine umschließende Transaktion geschehen würde).

Das bedeutet:

• wir beginnen mit einer langwierigen Transaktion
• TRUNCATE die AccessExclusive-Sperre
• wir machen lange Einfügungen, während alle anderen in dieser Zeit nicht einmal SELECT

Es läuft etwas schief…

ALTER TABLE… RENAME… / DROP TABLE …

Eine Möglichkeit wäre, alles in eine separate neue Tabelle einzufügen und sie dann einfach in die alte umzubenennen. Zwei unangenehme Kleinigkeiten:

• das ist auch AccessExclusive, obwohl es deutlich weniger Zeit in Anspruch nimmt
• alle Abfragepläne/Statistiken dieser Tabelle werden zurückgesetzt, man muss ANALYZE durchführen
• alle Fremdschlüssel (FK) auf die Tabelle brechen

Es gab einen WIP-Patch von Simon Riggs, der vorschlug, die ALTER-Operation für den Austausch des Tabellenkörpers auf Dateiebene durchzuführen, ohne die Statistiken und FK zu beeinträchtigen, aber er konnte keine Mehrheit bilden.

DELETE, UPDATE, INSERT

Also entscheiden wir uns für die nicht-blockierende Variante aus drei Operationen. Fast drei… Wie machen wir das am effektivsten?

-- Wir führen alles innerhalb der Transaktion durch, sodass niemand die "Zwischenzustände" sieht
BEGIN;

-- Wir erstellen eine temporäre Tabelle mit den 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, schnell den neuen Datensatz über COPY einfügen
COPY tmp FROM STDIN;
-- ...
-- .

-- Entfernen der fehlenden Einträge
DELETE FROM
  dst D
USING
  dst X
LEFT JOIN
  tmp Y
    USING(pk1, pk2) -- primärschlüssel Felder
WHERE
  (D.pk1, D.pk2) = (X.pk1, X.pk2) AND
  Y IS NOT DISTINCT FROM NULL; -- "Anti-Join"

-- Aktualisieren der verbleibenden
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); -- keine Notwendigkeit, übereinstimmende Einträge zu aktualisieren

-- Einfügen der fehlenden Einträge
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. Nachbearbeitung des Imports

Im selben KLR müssen alle geänderten Einträge zusätzlich durch die Nachbearbeitung geleitet werden – normalisieren, Schlüsselwörter extrahieren, in die benötigten Strukturen bringen. Aber wie erfahren wir – was genau geändert wurde, ohne dabei den Synchronisierungscode zu komplizieren, idealerweise ihn überhaupt nicht zu berühren?

Wenn der Schreibzugriff zum Zeitpunkt der Synchronisierung nur für Ihren Prozess verfügbar ist, können Sie einen Trigger verwenden, der uns alle Änderungen sammelt:

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

-- Tabellen mit der Historie der Änderungen
CREATE TABLE kladr$log(
  ro kladr, -- hier liegen die kompletten Abbildungen der alten/neuen Datensätze
  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üfung der Notwendigkeit der Protokollierung bei Aktualisierung des Datensatzes
  IF TG_OP = 'UPDATE' THEN
    IF NEW IS NOT DISTINCT FROM OLD THEN
      RETURN NEW;
    END IF;
  END IF;
  -- Erstellung des Protokolleintrags
  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 vor Beginn der Synchronisierung die Trigger 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();

Dann extrahieren wir in Ruhe alle benötigten Änderungen aus den Log-Tabellen und führen diese durch zusätzliche Verarbeiter.

3.3. Import von verknüpften Datensätzen

Zuvor haben wir Fälle betrachtet, in denen die Datenstrukturen von Quelle und Ziel übereinstimmen. Aber was tun, wenn der Export aus einem externen System ein Format hat, das von unserer Datenbankstruktur abweicht?

Nehmen wir als Beispiel die Speicherung von Kunden und deren Rechnungen, das klassische „Viele-zu-eins“-Szenario:

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

Der Export aus der externen Quelle kommt in Form von „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önnten die Kundendaten in dieser Form dupliziert werden, wobei der Hauptdatensatz die „Rechnung“ ist:

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

Für das Modell fügen wir einfach unsere Testdaten ein, aber denken Sie daran — COPY es ist effektiver!

INSERT INTO invoice_import
VALUES
  ('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);

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

CREATE TEMPORARY TABLE client_import AS
SELECT DISTINCT ON(client_inn)
-- man kann einfach SELECT DISTINCT verwenden, wenn die Daten von vornherein nicht widersprüchlich 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 zuerst ermitteln oder generieren. Fügen wir dafür die entsprechenden Felder hinzu:

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

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

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

-- wir fügen fehlende Datensätze ein und tragen deren IDs ein
WITH ins AS (
  INSERT INTO client(
    inn
  , name
  )
  SELECT
    inn
  , name
  FROM
    client_import
  WHERE
    client_id IS NULL -- wenn die ID nicht eingetragen werden konnte
  RETURNING *
)
UPDATE
  client_import T
SET
  client_id = D.client_id
FROM
  ins D
WHERE
  T.inn = D.inn; -- eindeutiger Schlüssel

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

Eigentlich ist das alles — invoice_import nun haben wir das Verknüpfungsfeld ausgefüllt client_id, mit dem wir die Rechnung einfügen werden.

Quelle: habr.com