Einrichten von Openflow Connector for PostgreSQL¶

Wal_level konfigurieren¶

Openflow Connector for PostgreSQL erfordert, dass wal_level auf logical gesetzt wird.

Je nachdem, wo Ihr PostgreSQL-Server gehostet wird, können Sie den wal_level wie folgt konfigurieren:

Eine Publikation erstellen¶

Openflow Connector for PostgreSQL erfordert, dass eine -Veröffentlichung erstellt und in PostgreSQL konfiguriert wird, bevor die Replikation beginnt. Sie können es für alle oder eine Teilmenge von Tabellen sowie für bestimmte Tabellen mit nur bestimmten Spalten erstellen. Vergewissern Sie sich, dass alle Tabellen und Spalten, die Sie replizieren möchten, in der Veröffentlichung enthalten sind. Sie können die Veröffentlichung auch später ändern, während der Konnektor läuft. Um eine Publikation zu erstellen und zu konfigurieren, gehen Sie wie folgt vor:

1. Melden Sie sich als Benutzer mit der an CREATE-Berechtigung für die Datenbank und führen Sie die folgende Abfrage aus:

   CREATE PUBLICATION <publication name>;
   

   Copy

2. Definieren Sie Tabellen, die der Datenbankagent sehen kann:

ALTER PUBLICATION <publication name> ADD TABLE <table name>;

Copy

Wichtig

PostgreSQL 15 und später unterstützt die Konfiguration von Veröffentlichungen für eine bestimmte Untergruppe von Tabellenspalten. Damit der Konnektor dies korrekt unterstützt, müssen Sie die Spaltenfiltereinstellungen verwenden, um die gleichen Spalten einzuschließen, die auch in der Veröffentlichung eingestellt sind.

Ohne diese Einstellung verhält sich der Konnektor wie folgt:

• In der Zieltabelle werden die Spalten, die nicht im Filter enthalten sind, mit dem Suffix __DELETED versehen. Alle Daten, die während der Snapshot-Phase repliziert wurden, werden beibehalten.

• Nachdem Sie der Veröffentlichung neue Spalten hinzugefügt haben, schlägt die Tabelle dauerhaft fehl und Sie müssen ihre Replikation neu starten.

Weitere Informationen dazu finden Sie unter ALTER PUBLICATION.

Replizieren Sie eine Reihe von Tabellen in Echtzeit¶

1. Klicken Sie mit der rechten Maustaste auf die importierte Prozessgruppe und wählen Sie Parameters.

2. Geben Sie die erforderlichen Parameterwerte ein, wie unter Ablaufparameter beschrieben.

Replizieren von Tabellen von einem PostgreSQL Replikatserver¶

Der Konnektor kann Daten von einem primären Server, einem Hot-Standby-Replikat oder einem Abonnentenserver mit logischer Replikation erfassen. Bevor Sie den Konnektor für die Verbindung mit einem PostgreSQL Replikat konfigurieren, stellen Sie sicher, dass die Replikation zwischen Primär- und Replikatknoten korrekt funktioniert. Wenn Sie Probleme mit fehlenden Daten im Konnektor untersuchen, stellen Sie zunächst sicher, dass auf dem vom Konnektor verwendeten Server fehlende Zeilen vorhanden sind.

Zusätzliche Hinweise beim Herstellen einer Verbindung mit einem Standby-Replikat:

• Es wird nur die Verbindung zu einem Hot-Standby-Replikat unterstützt. Beachten Sie, dass Snowflake-Standardreplikate erst dann Verbindungen von Clients akzeptieren können, wenn diese zu einer Primärinstanz heraufgestuft wurden.

• PostgreSQL-Version des Servers muss >= 16 sein.

• Die Veröffentlichung, die vom Konnektor benötigt wird, muss auf dem primären Server erstellt werden, nicht auf dem Standby-Server. Der Standby-Server ist schreibgeschützt und erlaubt das Erstellen einer Veröffentlichung nicht.

Wenn Sie eine Verbindung zu einer Hot-Standby-Instanz herstellen und eine Zeitüberschreitung bei Es wird versucht, den Replikations-Slot zu erstellen<replication slot> sehen. Wenn Sie eine Verbindung zu einer Standby-Instanz herstellen, stellen Sie sicher, dass auf der primären PostgreSQL-Instanz etwas Datenverkehr vorhanden ist, da der Aufruf zum Erstellen eines Replikation-Slots sonst nie zurückgegeben wird. Fehler im Openflow-Bundle oder der Read PostgreSQL CDC Stream-Prozessor startet nicht, melden Sie sich bei der primären PostgreSQL-Instanz an und führen Sie die folgende Abfrage aus:

SELECT pg_log_standby_snapshot();

Der Fehler tritt auf, wenn es keine Datenänderungen auf dem primären Server gibt. So kann der Konnektor installiert werden, während ein Replikations-Slot auf dem Server erstellt wird. Das liegt daran, dass der Replikatserver Informationen über die Ausführung von Transaktionen vom Server benötigt, um einen Replikations-Slot erstellen zu können. Primärserver senden die Informationen nicht, wenn sie im Leerlauf sind. Die Funktion pg_log_standby_snapshot() zwingt den primären Server, Informationen über ausgeführte Transaktionen an den Replikatserver zu senden.

Entfernen und erneutes Hinzufügen einer Tabelle zur Replikation¶

Um eine Tabelle aus der Replikation zu entfernen, stellen Sie sicher, dass sie aus den Parametern Included Table Names oder Included Table Regex im Replikationsparameterkontext entfernt wird.

Wenn Sie die Tabelle später wieder zur Replikation hinzufügen möchten, löschen Sie zunächst die entsprechende Zieltabelle in Snowflake. Danach fügen Sie die Tabelle wieder zu den Parametern Included Table Names oder Included Table Regex hinzu. Dadurch wird sichergestellt, dass der Replikationsprozess für die Tabelle neu beginnt.

Diese Vorgehensweise kann auch zur Wiederherstellung nach einer fehlgeschlagenen Tabellenreplikation verwendet werden.

Replizieren einer Teilmenge von Spalten in einer Tabelle¶

Der Konnektor kann die replizierten Daten pro Tabelle auf eine Teilmenge der konfigurierten Spalten filtern.

Um Filter auf Spalten anzuwenden, ändern Sie die Eigenschaft „Column Filter“ im Replikationsparameterkontext und fügen Sie ein Array mit Konfigurationen hinzu, wobei Sie für jede Tabelle, auf die Sie einen Filter anwenden möchten, einen Eintrag hinzufügen.

Spalten können nach Name oder Muster einbezogen oder ausgeschlossen werden. Sie können eine einzelne Bedingung pro Tabelle anwenden oder mehrere Bedingungen kombinieren, wobei Ausschlüsse immer Vorrang vor Einbeziehungen haben.

Das folgende Beispiel zeigt die verfügbaren Felder. schema und table sind obligatorisch, und dann ist eines oder mehrere der Felder included, excluded, includedPattern, excludedPattern erforderlich.

[
    {
        "schema": "<source table schema>",
        "table" : "<source table name>",
        "included": ["<column name>", "<column name>"],
        "excluded": ["<column name>", "<column name>"],
        "includedPattern": "<regular expression>",
        "excludedPattern": "<regular expression>",
    }
]

Verfolgen von Datenänderungen in Tabellen¶

Der Konnektor repliziert nicht nur den aktuellen Zustand der Daten aus den Quelltabellen, sondern auch jeden Zustand jeder Zeile aus jedem Änderungssatz. Diese Daten werden in Journaltabellen gespeichert, die in demselben Schema wie die Zieltabelle erstellt wurden.

Die Journaltabellennamen haben folgendes Format: <source table name>_JOURNAL_<timestamp>_<schema generation> where <timestamp> is the value of epoch seconds when the source table was added to replication, and <schema generation> ist eine ganze Zahl, die mit jeder Schemaänderung in der Quelltabelle erhöht wird. Infolgedessen haben Quelltabellen, die Schemaänderungen unterliegen, mehrere Journaltabellen.

Wenn eine Tabelle aus der Replikation entfernt und dann wieder hinzugefügt wird, ändert sich der Wert von <timestamp> und <schema generation> beginnt wieder bei 1.

Der Konnektor löscht nie Journaltabellen, sondern verwendet das neueste Journal für jede replizierte Quelltabelle und liest nur Nur-Anfügen-Streams über Journale. Um den Speicher wieder freizugeben, können Sie Folgendes tun:

• Sie können alle Journaltabellen jederzeit kürzen.

• Löschen Sie die Journaltabellen, die sich auf Quelltabellen beziehen, die aus der Replikation entfernt wurden.

• Löschen Sie alle Journaltabellen bis auf die neueste Generation aktiv replizierter Tabellen.

Wenn Ihr Konnektor beispielsweise so eingestellt ist, dass er die Quelltabelle orders aktiv repliziert, und Sie zuvor die Tabelle customers aus der Replikation entfernt haben, haben Sie möglicherweise die folgenden Journaltabellen. In diesem Fall können Sie alle außer orders_5678_2 löschen.

customers_1234_1
customers_1234_2
orders_5678_1
orders_5678_2

Planung von Zusammenführungsaufgaben konfigurieren¶

Der Konnektor verwendet ein Warehouse, um Daten aus der Änderungsdatenerfassung (CDC) in Zieltabellen zusammenzuführen. Diese Operation wird durch den Prozessor MergeSnowflakeJournalTable ausgelöst. Wenn es keine neuen Änderungen gibt oder wenn keine neuen FlowFiles in der MergeSnowflakeJournalTable-Warteschlange warten, wird keine Zusammenführung ausgelöst und das Warehouse wird automatisch ausgesetzt.

Um die Warehouse-Kosten zu begrenzen und die Zusammenführungen nur auf die geplante Zeit zu beschränken, verwenden Sie den CRON-Ausdruck im Parameter „Merge task Schedule CRON“. Er drosselt die an den MergeSnowflakeJournalTable-Prozessor gelangenden FlowFiles und die Zusammenführung wird nur in einem bestimmten Zeitraum ausgelöst. Weitere Informationen zur Zeitplanung finden Sie unter Zeitplanungsstrategie.

Den Konnektor anhalten oder löschen¶

Wenn Sie den Konnektor anhalten oder entfernen, müssen Sie den Replikationsslot berücksichtigen, den der Konnektor verwendet.

Der Konnektor erstellt einen eigenen Replikationsslot mit einem Namen, der mit snowflake_connector_ beginnt, gefolgt von einem zufälligen Suffix. Wenn der Konnektor den Replikationsstrom liest, schiebt er den Slot weiter, sodass PostgreSQL sein WAL-Protokoll kürzen und Speicherplatz freigeben kann.

Wenn der Konnektor pausiert, wird der Slot nicht erweitert, und Änderungen an der Quelldatenbank erhöhen weiterhin die WAL-Protokollgröße. Sie sollten den Konnektor nicht für längere Zeit pausieren lassen, insbesondere nicht bei Datenbanken mit hohem Datenverkehr.

Wenn der Konnektor entfernt wird, sei es durch Löschen aus dem Openflow-Canvas oder auf andere Weise, z. B. durch Löschen der gesamten Openflow-Instanz, bleibt der Replikationsslot bestehen und muss manuell gelöscht werden.

Wenn Sie mehrere Konnektorinstanzen haben, die von derselben PostgreSQL-Datenbank replizieren, erstellt jede Instanz ihren eigenen, eindeutig benannten Replikationsslot. Wenn Sie einen Replikationsslot manuell löschen, vergewissern Sie sich, dass es sich um den richtigen handelt. Sie können sehen, welcher Replikationsslot von einer bestimmten Konnektorinstanz verwendet wird, indem Sie den Status des CaptureChangePostgreSQL-Prozessors überprüfen.