Verwalten von PostgreSQL-Aufnahmepipelines

Auf dieser Seite werden die laufenden Vorgänge für die Wartung von PostgreSQL-Aufnahmepipelines beschrieben.

Allgemeine Pipelinewartung

Die Pipelinewartungsaufgaben in diesem Abschnitt gelten für alle verwalteten Connectors in Lakeflow Connect.

Allgemeine Pipelinewartungsaufgaben finden Sie unter Allgemeine Pipelinewartungsaufgaben.

Entfernen nicht verwendeter Stagingdateien

Bei Aufnahmepipelines, die Sie nach dem 6. Januar 2025 erstellen, werden Volume-Stagingdaten nach 25 Tagen automatisch für die Löschung geplant und nach 30 Tagen physisch entfernt. Eine Aufnahmepipeline, die seit 25 Tagen oder länger nicht erfolgreich abgeschlossen ist, kann zu Datenlücken in den Zieltabellen führen. Um Lücken zu vermeiden, lösen Sie eine vollständige Aktualisierung der Zieltabellen aus.

Für vor dem 6. Januar 2025 erstellte Erfassungspipelines wenden Sie sich an den Databricks-Support, um die manuelle Aktivierung der automatischen Aufbewahrungsverwaltung für das Staging von CDC-Daten anzufordern.

Die folgenden Daten werden automatisch bereinigt:

• CDC-Datendateien
• Momentaufnahmedateien
• Stagingtabellendaten

Anschlussspezifische Pipeline-Wartung

Die Pipelinewartungsaufgaben in diesem Abschnitt sind spezifisch für den PostgreSQL-Connector.

Hinzufügen neuer Tabellen zur Replikation

So fügen Sie einem vorhandenen Replikationsfluss neue Tabellen hinzu:

1. Erteilen Sie dem Replikationsbenutzer die erforderlichen Berechtigungen. Eine vollständige Liste der erforderlichen Berechtigungen finden Sie unter Den Benutzeranforderungen der PostgreSQL-Datenbank.

2. Legen Sie die Replikatidentität für die neuen Tabellen basierend auf ihrer Struktur fest. Informationen zum Auswählen der richtigen Replikatidentitätseinstellung finden Sie unter "Festlegen der Replikatidentität für Tabellen ".

3. Fügen Sie der Publikation die Tabellen hinzu:

   ALTER PUBLICATION databricks_publication ADD TABLE schema_name.new_table;
   

4. Aktualisieren Sie die Konfiguration der Aufnahmepipeline so, dass sie die neuen Tabellen enthält. Dazu können Sie die Benutzeroberfläche von Azure Databricks verwenden oder das ingestion_definition in Ihren Databricks Asset Bundles oder im CLI-Befehl aktualisieren.

5. Starten Sie das Ingestions-Gateway neu, um die neuen Tabellen zu entdecken. Das Gateway sucht regelmäßig nach neuen Tabellen, aber durch den Neustart des Gateways wird der Ermittlungsprozess beschleunigt.

Replikationsslots bereinigen

Wenn Sie eine Aufnahmepipeline löschen, ** wird der Replikationsplatz nicht automatisch aus der Quell-PostgreSQL-Datenbank **entfernt. Nicht verwendete Replikationsplätze können dazu führen, dass sich Write-Ahead Log-Dateien (WAL) ansammeln und möglicherweise Speicherplatz in der Quelldatenbank auffüllen.

So listen Sie alle Replikationsplätze auf:

SELECT slot_name, slot_type, active, restart_lsn, pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained_wal
FROM pg_replication_slots;

So legen Sie einen Replikationsplatz ab, der nicht mehr benötigt wird:

SELECT pg_drop_replication_slot('slot_name');

Bereinigung der Inline-DDL-Verfolgung

Wenn Sie die Inline-DDL-Nachverfolgung deaktivieren, führen Sie die folgenden Schritte für jede Datenbank aus, um vom Überwachungsskript erstellte Objekte zu bereinigen.

1. Löschen Sie die Ereignistrigger:

   DROP EVENT TRIGGER IF EXISTS lakeflow_ddl_trigger CASCADE;
   

2. Entfernen Sie die Überwachungstabelle aus der Publikation:

   ALTER PUBLICATION databricks_publication DROP TABLE public.lakeflow_ddl_audit;
   

3. Ablegen der Überwachungstabelle:

   DROP TABLE IF EXISTS public.lakeflow_ddl_audit CASCADE;
   

Überwachen von Replikations-Slots

Überwachen Sie den Status der Replikationsplätze, um sicherzustellen, dass sie aktiv sind und WAL-Daten verbrauchen:

SELECT slot_name,
       active,
       wal_status,
       active_pid,
       restart_lsn,
       confirmed_flush_lsn,
       pg_current_wal_lsn() AS current_lsn,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS replication_lag
FROM pg_replication_slots
WHERE slot_name LIKE 'databricks%';

Große Replikationsverzögerungswerte können auf eines der folgenden Probleme hinweisen:

• Das Gateway zur Datenaufnahme kann die Änderungen in der Quelldatenbank nicht nachvollziehen.
• Das Ingestion-Gateway wurde für einen längeren Zeitraum gestoppt.
• Netzwerkkonnektivitätsprobleme zwischen dem Gateway und der Quelldatenbank.

Wenn ein Replikationsplatz inaktiv (active = false) ist und Sie bestätigt haben, dass die entsprechende Pipeline nicht mehr benötigt wird, legen Sie den Replikationsplatz ab, um die Ressourcen freizugeben. Siehe Replikations-Slots bereinigen.

Überwachung der WAL-Datenträgernutzung

Überwachen Sie die Write-Ahead-Log (WAL)-Datenträgerauslastung, um Festplattenspeicherprobleme zu vermeiden.

SELECT pg_size_pretty(sum(size)) AS wal_size
FROM pg_ls_waldir();

Um die WAL-Aufbewahrung für einen bestimmten Replikations-Slot zu überprüfen:

SELECT slot_name,
       active,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained_wal,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) AS pending_wal
FROM pg_replication_slots
WHERE slot_name = 'your_slot_name';

Wenn die WAL-Datenträgerauslastung hoch ist, führen Sie die folgenden Schritte aus:

1. Stellen Sie sicher, dass das Ingestion-Gateway kontinuierlich ausgeführt wird.

2. Überprüfen Sie die Gatewayprotokolle auf Fehler, die verhindern könnten, dass es WAL-Daten verarbeitet.

3. Erwägen Sie die Einstellung max_slot_wal_keep_size zum Einschränken der WAL-Aufbewahrung (PostgreSQL 13 oder höher):

   ALTER SYSTEM SET max_slot_wal_keep_size = '10GB';
   SELECT pg_reload_conf();
   

   Warnung

   Die Einstellung max_slot_wal_keep_size kann dazu führen, dass Replikationsplätze ungültig werden, wenn der WAL-Aufbewahrungsgrenzwert überschritten wird und eine vollständige Aktualisierung aller Tabellen erforderlich ist.

Starten Sie das Ingestion-Gateway neu

Um die Belastung der Quelldatenbank zu verringern, prüft das Ingestion-Gateway nur noch periodisch auf neue Tabellen. Es kann bis zu 6 Stunden dauern, bis das Gateway neue Tabellen ermittelt. Wenn Sie diesen Prozess beschleunigen möchten, starten Sie das Gateway neu.

Starten Sie das Gateway in den folgenden Situationen neu:

• Sie haben Konfigurationsänderungen an der Quelldatenbank vorgenommen.
• Beim Gateway treten Fehler oder Leistungsprobleme auf.

Aktualisieren von Publikationen

Wenn Sie ändern müssen, welche Tabellen in der Replikation enthalten sind:

-- Add a table to the publication
ALTER PUBLICATION databricks_publication ADD TABLE schema_name.table_name;

-- Remove a table from the publication
ALTER PUBLICATION databricks_publication DROP TABLE schema_name.table_name;

-- List all tables in a publication
SELECT schemaname, tablename
FROM pg_publication_tables
WHERE pubname = 'databricks_publication';

Starten Sie nach dem Aktualisieren der Veröffentlichung das Ingestions-Gateway neu, um die Änderungen anzuwenden.