Maintenir les pipelines d'ingestion PostgreSQL

info

Aperçu

Le connecteur PostgreSQL pour Lakeflow Connect est en aperçu public. Contactez votre équipe de compte Databricks pour vous inscrire à l'aperçu public.

Cette page décrit les opérations continues pour la maintenance des pipelines d'ingestion PostgreSQL.

Maintenance générale du pipeline

Les tâches de maintenance de pipeline de cette section s'appliquent à tous les connecteurs gérés dans Lakeflow Connect.

Pour les tâches générales de maintenance de pipeline, consultez Tâches courantes de maintenance de pipeline.

Supprimer les fichiers de staging non utilisés

Pour les pipelines d'ingestion que vous créez après le 6 janvier 2025, les données de staging de volume sont automatiquement planifiées pour être supprimées après 25 jours et physiquement retirées après 30 jours. Un pipeline d'ingestion qui n'a pas été complété avec succès pendant 25 jours ou plus peut entraîner des lacunes de données dans les tables de destination. Pour éviter les lacunes, Trigger une refresh complète des tables cibles.

Pour les pipelines d'ingestion créés avant le 6 janvier 2025, contactez le support Databricks pour demander l'activation manuelle de la gestion automatique de la rétention des données CDC de staging.

Les données suivantes sont automatiquement nettoyées :

Fichiers de données CDC
Fichiers instantanés
Données de table intermédiaires

Maintenance du pipeline spécifique au connecteur

Les tâches de maintenance du pipeline dans cette section sont spécifiques au connecteur PostgreSQL.

Ajouter de nouvelles tables à la réplication

Pour ajouter de nouvelles tables à un flux de réplication existant :

Accordez les privilèges nécessaires à l'utilisateur de réplication. Pour une liste complète des privilèges requis, voir les exigences relatives aux utilisateurs de la base de données PostgreSQL.
Définissez l'identité de réplication pour les nouvelles tables en fonction de leur structure. Consultez Définir l'identité du réplica pour les tables pour obtenir des conseils sur le choix du paramètre d'identité de réplica correct.

Ajoutez les tables à la publication :

SQL
ALTER PUBLICATION databricks_publication ADD TABLE schema_name.new_table;

Mettez à jour la configuration du pipeline d'ingestion pour inclure les nouvelles tables. Vous pouvez le faire via l'interface utilisateur de Databricks ou en mettant à jour le ingestion_definition dans votre bundle Declarative Automation Bundles ou votre commande CLI.
Redémarrez la passerelle d'ingestion pour découvrir les nouvelles tables. La passerelle vérifie périodiquement les nouvelles tables, mais le redémarrage de la passerelle accélère le processus de découverte.

Nettoyez les emplacements de réplication

Lorsque vous supprimez un pipeline d'ingestion, l'emplacement de réplication n'est pas automatiquement supprimé de la base de données PostgreSQL source.

important

Les emplacements de réplication inutilisés peuvent entraîner l'accumulation de fichiers WAL (Write-Ahead Log), remplissant potentiellement l'espace disque de la base de données source.

Pour lister tous les emplacements de réplication :

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

Pour supprimer un emplacement de réplication qui n'est plus nécessaire, vous devez être connecté en tant qu'utilisateur disposant du privilège REPLICATION. Si vous êtes connecté en tant que super-utilisateur ou administrateur, basculez d'abord vers l'utilisateur de réplication :

SQL
SET ROLE databricks_replication;
SELECT pg_drop_replication_slot('databricks_slot');
RESET ROLE;

Nettoyer le suivi DDL en ligne

Si vous désactivez le suivi DDL intégré, exécutez les étapes ci-dessous pour chaque base de données afin de nettoyer les objets créés par le script d'audit. Les noms d'objet incluent un suffixe de version (actuellement _1_0) qui correspond à la version du script utilisée lors de la configuration.

Supprimez les Trigger d'événement :

SQL
DROP EVENT TRIGGER IF EXISTS lakeflow_ddl_audit_trigger_1_0 CASCADE;
DROP EVENT TRIGGER IF EXISTS lakeflow_drop_ddl_audit_trigger_1_0 CASCADE;

Supprimer la table d'audit de la publication :

SQL
ALTER PUBLICATION databricks_publication DROP TABLE public.lakeflow_ddl_audit_table_1_0;

Supprimez les fonctions d'audit :

SQL
DROP FUNCTION IF EXISTS public.lakeflow_ddl_audit_function_1_0() CASCADE;
DROP FUNCTION IF EXISTS public.lakeflow_drop_ddl_audit_function_1_0() CASCADE;

Supprimer la table d'audit :

SQL
DROP TABLE IF EXISTS public.lakeflow_ddl_audit_table_1_0 CASCADE;

Surveiller les emplacements de réplication

Surveillez l'état des emplacements de réplication pour vous assurer qu'ils sont actifs et consomment des données WAL :

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

Des valeurs de décalage de réplication élevées peuvent indiquer l'un des problèmes suivants :

La passerelle d’ingestion ne parvient pas à suivre les modifications de la base de données source.
La passerelle d'ingestion a été arrêtée pendant une période prolongée.
Problèmes de connectivité réseau entre la passerelle et la base de données source.

Si un emplacement de réplication est inactif (active = false) et que vous avez confirmé que le pipeline correspondant n'est plus nécessaire, supprimez l'emplacement de réplication pour libérer les ressources. Consultez Nettoyage des emplacements de réplication.

Surveiller l'utilisation du disque WAL

Surveillez l’utilisation du disque du Write-Ahead Log (WAL) pour éviter les problèmes d’espace disque :

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

Pour vérifier la rétention WAL pour un emplacement de réplication spécifique :

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

remarque

Si max_slot_wal_keep_size est correctement configuré lors de la configuration de la source (comme recommandé dans Limiter la rétention des WAL pour les emplacements de réplication), les emplacements de réplication inactifs ne provoqueront pas une croissance illimitée des WAL. L'emplacement sera invalidé lorsque la limite sera atteinte, ce qui empêchera les plantages de la base de données.

Si l'utilisation du disque WAL est élevée, effectuez les étapes suivantes :

Vérifiez que la passerelle d'ingestion fonctionne en continu.
Vérifiez les Logs de la passerelle pour les erreurs qui pourraient l'empêcher de consommer les données WAL.
Envisagez de définir max_slot_wal_keep_size pour limiter la rétention WAL (PostgreSQL 13 ou version ultérieure) :
SQL
```
ALTER SYSTEM SET max_slot_wal_keep_size = '10GB';
SELECT pg_reload_conf();
```

attention

Le paramètre max_slot_wal_keep_size peut entraîner l'invalidation des emplacements de réplication si la limite de rétention WAL est dépassée, nécessitant un full refresh de toutes les tables.

Redémarrer la passerelle d’ingestion

Afin de réduire la charge sur la base de données source, la passerelle d'ingestion ne vérifie périodiquement que les nouvelles tables. La passerelle peut prendre jusqu'à 6 heures pour découvrir de nouvelles tables. Si vous souhaitez accélérer ce processus, redémarrez la passerelle.

De plus, redémarrez la passerelle dans les situations suivantes :

Vous avez apporté des modifications de configuration à la base de données source.
La passerelle rencontre des erreurs ou des problèmes de performance.

Mettre à jour les publications

Si vous devez modifier les tables incluses dans la réplication :

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

Après la mise à jour de la publication, redémarrez la passerelle d'ingestion pour appliquer les modifications.

Supprimer tous les objets Lakeflow Connect de la base de données source

Si vous n'avez plus besoin de répliquer à partir d'une base de données PostgreSQL, exécutez les étapes suivantes pour supprimer tous les objets créés lors de la configuration de la source. Exécutez ces commandes en tant que super-utilisateur ou propriétaire de la table, sauf indication contraire.

Si le suivi DDL en ligne a été configuré, nettoyez-le d'abord. Voir Nettoyer le suivi DDL en ligne.

Supprimez l'emplacement de réplication. Ceci requiert le privilège REPLICATION :

SQL
SET ROLE databricks_replication;
SELECT pg_drop_replication_slot('databricks_slot');
RESET ROLE;

Supprimer la publication :

SQL
DROP PUBLICATION IF EXISTS databricks_publication;

Révoquer les privilèges et supprimer l’utilisateur de réplication :

SQL
REVOKE ALL PRIVILEGES ON ALL TABLES IN SCHEMA schema_name FROM databricks_replication;
REVOKE USAGE ON SCHEMA schema_name FROM databricks_replication;
REVOKE CONNECT ON DATABASE your_database FROM databricks_replication;
DROP USER IF EXISTS databricks_replication;

Maintenance générale du pipeline​

Supprimer les fichiers de staging non utilisés​

Maintenance du pipeline spécifique au connecteur​

Ajouter de nouvelles tables à la réplication​

Nettoyez les emplacements de réplication​

Nettoyer le suivi DDL en ligne​

Surveiller les emplacements de réplication​

Surveiller l'utilisation du disque WAL​

Redémarrer la passerelle d’ingestion​

Mettre à jour les publications​

Supprimer tous les objets Lakeflow Connect de la base de données source​