Obsługa potoków pozyskiwania bazy danych PostgreSQL

Na tej stronie opisano bieżące operacje obsługi potoków pozyskiwania bazy danych PostgreSQL.

Konserwacja ogólna rurociągu

Zadania konserwacji ciągu technologicznego w tej sekcji mają zastosowanie do wszystkich zarządzanych łączników w programie Lakeflow Connect.

Ogólne zadania konserwacji potoku można znaleźć w temacie Typowe zadania konserwacji potoku.

Usuwanie nieużywanych plików przejściowych

W przypadku potoków pozyskiwania utworzonych po 6 stycznia 2025 r. dane etapowania wolumenu są automatycznie przeznaczane do usunięcia po 25 dniach i fizycznie usunięte po 30 dniach. Potok przetwarzania danych, który nie został pomyślnie ukończony od 25 dni lub dłużej, może spowodować braki danych w tabelach docelowych. Aby uniknąć przerw, uruchom pełne odświeżanie tabel docelowych.

W przypadku potoków danych utworzonych przed 6 stycznia 2025 r. skontaktuj się z pomocą techniczną Databricks, aby poprosić o ręczne włączenie automatycznego zarządzania retencją dla danych CDC umieszczanych w etapie przetwarzania wstępnego.

Następujące dane są automatycznie czyszczone:

• Pliki danych CDC
• Pliki migawek
• Dane tabeli przejściowej

Konserwacja procesu specyficznego dla łącznika

Zadania konserwacji przepływu danych w tej sekcji są specyficzne dla łącznika PostgreSQL.

Dodawanie nowych tabel do replikacji

Aby dodać nowe tabele do istniejącego przepływu replikacji:

1. Przyznaj użytkownikowi replikacji niezbędne uprawnienia. Aby uzyskać pełną listę wymaganych uprawnień, zobacz Wymagania użytkownika bazy danych PostgreSQL.

2. Ustaw tożsamość repliki dla nowych tabel na podstawie ich struktury. Zobacz Ustawianie tożsamości repliki dla tabel, aby uzyskać wskazówki dotyczące wybierania poprawnego ustawienia tożsamości repliki.

3. Dodaj tabele do publikacji:

   ALTER PUBLICATION databricks_publication ADD TABLE schema_name.new_table;
   

4. Zaktualizuj konfigurację potoku przetwarzania danych, aby uwzględnić nowe tabele. Możesz to zrobić przy użyciu Azure Databricks UI lub aktualizując ingestion_definition w pakiecie zasobów Databricks albo za pomocą polecenia wiersza poleceń CLI.

5. Uruchom ponownie bramę wprowadzania, aby odkryć nowe tabele. Brama regularnie sprawdza nowe tabele, ale ponowne uruchomienie bramy przyspiesza proces odkrywania nowych tabel.

Czyszczenie miejsc replikacji

Po usunięciu potoku wprowadzania danych, ** gniazdo replikacji nie będzie automatycznie usunięte ze źródłowej bazy danych PostgreSQL **. Nieużywane miejsca replikacji mogą powodować gromadzenie plików dziennika Write-Ahead (WAL), co może potencjalnie spowodować wypełnienie miejsca na dysku w źródłowej bazie danych.

Aby wyświetlić listę wszystkich miejsc replikacji:

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;

Aby usunąć miejsce replikacji, które nie jest już potrzebne:

SELECT pg_drop_replication_slot('slot_name');

Wyczyść wbudowane śledzenie DDL

Jeśli wyłączysz wbudowane śledzenie DDL, uruchom poniższe kroki dla każdej bazy danych, aby wyczyścić obiekty utworzone przez skrypt inspekcji.

1. Usuń wyzwalacze zdarzeń

   DROP EVENT TRIGGER IF EXISTS lakeflow_ddl_trigger CASCADE;
   

2. Usuń tabelę inspekcji z publikacji:

   ALTER PUBLICATION databricks_publication DROP TABLE public.lakeflow_ddl_audit;
   

3. Upuść tabelę audytu.

   DROP TABLE IF EXISTS public.lakeflow_ddl_audit CASCADE;
   

Monitorowanie miejsc replikacji

Monitoruj stan miejsc replikacji, aby upewnić się, że są aktywne i zużywają dane WAL:

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%';

Duże wartości opóźnienia replikacji mogą wskazywać na jeden z następujących problemów:

• Brama wprowadzania danych nie nadąża za zmianami w źródłowej bazie danych.
• Brama pozyskiwania została zatrzymana przez dłuższy czas.
• Problemy z łącznością sieciową między bramą a źródłową bazą danych.

Jeśli slot replikacji jest nieaktywny (active = false) i potwierdzono, że powiązany potok nie jest już potrzebny, usuń slot replikacji, aby zwolnić zasoby. Zobacz Czyszczenie miejsc replikacji.

Monitorowanie użycia dysku WAL

Monitoruj użycie dysku dziennika Write-Ahead (WAL), aby zapobiec problemom z miejscem na dysku:

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

Aby sprawdzić przechowywanie pliku WAL dla określonego slotu replikacji:

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

Jeśli użycie dysku WAL jest wysokie, wykonaj następujące kroki:

1. Sprawdź, czy brama pozyskiwania jest uruchomiona w sposób ciągły.

2. Sprawdź dzienniki bramy pod kątem błędów, które mogą uniemożliwiać korzystanie z danych WAL.

3. Rozważ ustawienie max_slot_wal_keep_size w celu ograniczenia przechowywania WAL (PostgreSQL 13 lub nowszej wersji):

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

   Ostrzeżenie

   Ustawienie max_slot_wal_keep_size może spowodować unieważnienie slotów replikacji, jeśli przekroczono limit przechowywania WAL, co wymaga pełnego odświeżenia wszystkich tabel.

Uruchom ponownie bramę pozyskiwania

Aby zmniejszyć obciążenie źródłowej bazy danych, brama pozyskiwania sprawdza tylko okresowo nowe tabele. Może zająć do 6 godzin, aby brama odkryła nowe tabele. Jeśli chcesz przyspieszyć ten proces, uruchom ponownie bramę.

Ponadto uruchom ponownie bramę w następujących sytuacjach:

• Wprowadzono zmiany konfiguracji w źródłowej bazie danych.
• Brama napotyka błędy lub problemy z wydajnością.

Aktualizowanie publikacji

Jeśli musisz zmodyfikować tabele uwzględnione w replikacji:

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

Po zaktualizowaniu publikacji ponownie uruchom bramę przetwarzania danych, aby zmiany weszły w życie.