Einrichten von Openflow Connector for SQL Server¶

Ablaufparameter¶

Beginnen Sie mit der Einstellung der Parameter des SQLServer-Quellparameterkontexts, und fahren Sie dann mit dem SQLServer-Zielparameterkontext fort. Nachdem Sie dies abgeschlossen haben, aktivieren Sie den Konnektor. Der Konnektor stellt die Verbindung zu SQLServer und Snowflake her und startet dann mit der Ausführung. Der Konnektor repliziert jedoch erst dann Daten, wenn die zu replizierenden Tabellen explizit zu seiner Konfiguration hinzugefügt wurden.

Um bestimmte Tabellen für die Replikation zu konfigurieren, bearbeiten Sie die Aufnahmeparameter für SQLServer. Nachdem Sie die Änderungen am Kontext der Aufnahmeparameter für SQLServer vorgenommen haben, wird die Konfiguration vom Konnektor übernommen und der Replikationslebenszyklus für jede Tabelle gestartet.

Replizieren von Tabellen von einem Replikatserver von SQL Server¶

Der Konnektor kann Daten von einem Primärserver, von einem Abonnentenserver mithilfe der transaktionalen Replikation oder von einem sekundären Replikat in einer Always On-Verfügbarkeitsgruppe einlesen. Bevor Sie den Konnektor für die Verbindung mit einem SQL Server-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 Replikatserver fehlende Zeilen und Ereignisse der Änderungsverfolgung vorhanden sind.

Um die Kontinuität zu gewährleisten, stellen Sie sicher, dass derselbe Verbindungsbenutzer sowohl auf dem primären Server als auch auf dem Replikatserver verfügbar ist und Zugriff auf die Daten und die Änderungsverfolgungstabellen hat.

Always On-Verfügbarkeitsgruppen¶

Always On-Verfügbarkeitsgruppen sind eine hochverfügbare Lösung für die Notfallwiederherstellung, die synchronisierte Kopien von Datenbanken für Failover-Zwecke verwaltet. Der Konnektor kann aus einem sekundären Replikat in der Verfügbarkeitsgruppe lesen. Konfigurieren Sie für optimale Ergebnisse eine Überwachung für die Verfügbarkeitsgruppe, und verwenden Sie den DNS-Namen für die Überwachung im Parameter SQLServer Connection URL.

Replikation der Tabellen neu starten¶

Eine Tabelle in FAILED-Status – z. B. aufgrund eines fehlenden Primärschlüssels oder einer nicht unterstützten Schemaänderung – wird nicht automatisch neu gestartet. Wenn eine Tabelle in einen FAILED-Status übergeht oder Sie die Replikation von Grund auf neu starten müssen, gehen Sie wie folgt vor, um die Tabelle zu entfernen und zur Replikation erneut hinzuzufügen.

1. Die Tabelle aus den Ablaufparametern entfernen: Entfernen Sie im Kontext der Datenaufnahmeparameter entweder die Tabelle aus Included Table Names, oder ändern Sie Included Table Regex so, dass die Tabelle nicht mehr übereinstimmt.

2. Überprüfen, ob die Tabelle entfernt wurde:

   1. Klicken Sie in der Openflow-Laufzeitoberfläche mit der rechten Maustaste auf eine Prozessorgruppe, und wählen Sie Controller Services.

   2. Suchen Sie in der Tabelle mit den Controllerdiensten die Zeile Table State Store, klicken Sie auf die drei vertikalen Punkte rechts von der Zeile, und wählen Sie dann View State aus.

   Wichtig

   Sie müssen warten, bis der Status der Tabelle vollständig aus dieser Liste entfernt wurde, bevor Sie fortfahren können. Fahren Sie nicht fort, bis diese Konfigurationsänderung abgeschlossen ist.

3. Ziel bereinigen: Sobald der Status der Tabelle als „vollständig entfernt“ angezeigt wird, DROP Sie die Zieltabelle manuell in Snowflake. Beachten Sie, dass der Konnektor während der Snapshot-Phase keine vorhandene Zieltabelle überschreibt. Wenn die Tabelle noch existiert, schlägt die Replikation erneut fehl. Optional können die Journaltabelle und der Stream auch entfernt werden, wenn sie nicht mehr benötigt werden.

4. Fügen Sie die Tabelle erneut hinzu: Aktualisieren Sie den Parameter Included Table Names oder Included Table Regex so, dass die Tabelle wieder eingeschlossen wird.

5. Neustart überprüfen: Überprüfen Sie den Table State Store unter Verwendung der zuvor angegebenen Anweisungen. Der Status der Tabelle sollte mit dem Status NEW angezeigt werden, dann in SNAPSHOT_REPLICATION übergehen und schließlich in INCREMENTAL_REPLICATION.

Replizieren einer Teilmenge von Spalten in einer Tabelle¶

Der Konnektor filtert die pro Tabelle replizierten Daten in einer Teilmenge der konfigurierten Spalten.

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.

Schließen Sie Spalten nach Name oder Muster ein- und aus. Sie können eine Bedingung pro Tabelle anwenden oder mehrere Bedingungen kombinieren, wobei Ausschlüsse immer Vorrang vor Einschlüssen haben.

Das folgende Beispiel zeigt die Felder, die verfügbar sind. Die Felder schema und table sind Pflichtfelder. Eine oder mehrere der Optionen included, excluded, includedPattern, excludedPattern sind 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 den aktuellen Status der Daten aus den Quelltabellen sowie jeden Status jeder Zeile aus jedem Änderungssatz. Diese Daten werden in Journaltabellen gespeichert, die im gleichen Schema wie die Zieltabelle erstellt wurden.

Die Namen der Journaltabellen haben folgendes Format:<source_table_name>_JOURNAL_<timestamp>_<schema_generation>, wobei <timestamp> der Wert der Epochensekunden ist und angibt, wann die Quelltabelle zur Replikation hinzugefügt wurde, und <schema_generation> eine ganze Zahl ist, die mit jeder Schemaänderung in der Quelltabelle erhöht wird. Infolgedessen haben Quelltabellen, die Schemaänderungen unterliegen, mehrere Journaltabellen.

Wenn Sie eine Tabelle aus der Replikation entfernen und anschließend wieder hinzufügen, dann ändert sich der <timestamp>-Wert und``<schema_generation>`` beginnt wieder mit 1.

Der Konnektor löscht Journaltabellen nie, sondern verwendet das neueste Journal für jede replizierte Quelltabelle und liest dabei 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.

Verwenden Sie den CRON-Ausdruck im CRON-Parameter „Merge task Schedule“, um die Warehouse-Kosten zu begrenzen und die Zusammenführungen auf die geplanten Zeiten zu beschränken. Er drosselt die Flow-Dateien, die beim MergeSnowflakeJournalTable-Prozessor eingehen, und Zusammenführungen werden nur in einem bestimmten Zeitraum ausgelöst. Weitere Informationen zur Zeitplanung finden Sie unter Zeitplanungsstrategie.