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 Berechtigung CREATE in der Datenbank an und führen Sie folgende Abfrage aus:

CREATE PUBLICATION <publication name>;

Copy

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

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

Copy

Weitere Informationen dazu finden Sie unter ALTERPUBLICATION.

5. Als Snowflake-Kontoadministrator führen Sie die folgenden Aufgaben aus:

   1. Erstellen Sie einen Snowflake Benutzer mit dem Typ als SERVICE. Erstellen Sie eine Datenbank, um die replizierten Daten zu speichern, und richten Sie Berechtigungen für den Snowflake-Benutzer ein, um Objekte in dieser Datenbank zu erstellen, indem Sie die Berechtigungen USAGE und CREATE SCHEMA erteilen.

      CREATE DATABASE <destination_database>;
      CREATE USER <openflow_user> TYPE=SERVICE COMMENT='Service user for automated access of Openflow';
      CREATE ROLE <openflow_role>;
      GRANT ROLE <openflow_role> TO USER <openflow_user>;
      GRANT USAGE ON DATABASE <destination_database> TO ROLE <openflow_role>;
      GRANT CREATE SCHEMA ON DATABASE <destination_database> TO ROLE <openflow_role>;
      CREATE WAREHOUSE <openflow_warehouse>
           WITH
                WAREHOUSE_SIZE = 'MEDIUM'
                AUTO_SUSPEND = 300
                AUTO_RESUME = TRUE;
      GRANT USAGE, OPERATE ON WAREHOUSE <openflow_warehouse> TO ROLE <openflow_role>;
      

      Copy

   2. Erstellen Sie ein Paar sicherer Schlüssel (öffentlich und privat). Speichern Sie den privaten Schlüssel des Benutzers in einer Datei, die Sie der Konfiguration des Konnektors zur Verfügung stellen. Weisen Sie den öffentlichen Schlüssel dem Benutzer des Snowflake-Dienstes zu:

      ALTER USER <openflow_user> SET RSA_PUBLIC_KEY = 'thekey';
      

      Copy

      Weitere Informationen finden Sie unter Schlüsselpaar.

   3. Bestimmen Sie ein Warehouse, das der Konnektor verwenden soll. Beginnen Sie mit der Größe des MEDIUM-Warehouse und experimentieren Sie dann mit der Größe in Abhängigkeit von der Anzahl der zu replizierenden Tabellen und der Menge der übertragenen Daten. Große Tabellenzahlen lassen sich in der Regel besser mit Multi-Cluster-Warehouses als mit der Warehouse-Größe.

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.

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 Namen der Journaltabellen sind folgendermaßen formatiert: <source table name>_JOURNAL_<timestamp>_<schema generation>

wobei <timestamp> der Wert der Epochensekunden ist, als die Quelltabelle zur Replikation hinzugefügt wurde, und <schema generation> eine Ganzzahl ist, die mit jeder Schemaänderung der Quelltabelle steigt. Das bedeutet, dass eine Quelltabelle, die Schemaänderungen erfährt, mehrere Journaltabellen hat.

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 niemals Journaltabellen, sondern verwendet für jede replizierte Quelltabelle nur das aktuellste Journal. Wenn Sie den Speicherplatz wieder freigeben möchten, können Sie die Journaltabellen, die sich auf die aus der Replikation entfernten Quelltabellen beziehen, sowie alle außer den neuesten Generationen für aktiv replizierte Tabellen sicher löschen.

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.