Configurer PostgreSQL pour l'ingestion dans Databricks

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 tâches de configuration source pour l'ingestion de données de PostgreSQL vers Databricks à l'aide de Lakeflow Connect.

Identifiants utilisés lors de la configuration et de l'ingestion

L'ingestion PostgreSQL utilise deux ensembles de justificatifs différents à deux étapes différentes. Savoir quels identifiants utiliser où prévient les erreurs d'authentification et d'autorisation lors de la configuration.

Étape	Identifiants à utiliser	Pourquoi
Configuration de la source (cette page)	Un administrateur PostgreSQL, un superutilisateur ou le propriétaire de la table, connecté directement à la base de données source (par exemple, via psql ou la console de gestion de votre cloud provider).	La création de l'utilisateur de réplication, l'octroi de privilèges et la création de publications nécessitent des privilèges de superutilisateur ou de propriétaire de table que l'utilisateur de réplication ne possède pas. L'emplacement de réplication est créé par l'utilisateur de réplication lui-même, de sorte qu'un administrateur connecté à la base de données passe à ce rôle pour le créer. Pour la liste complète des privilèges, consultez les exigences relatives aux utilisateurs de bases de données PostgreSQL.
Pipeline de connexion et d'ingestion	L'utilisateur de réplication dédié (par exemple, databricks_replication) que vous créez lors de la configuration de la source.	La passerelle d'ingestion s'authentifie auprès de PostgreSQL en tant qu'utilisateur de réplication pour lire les modifications. Vous saisissez ces informations d'identification lorsque vous créez la connexion Unity Catalog. Voir Créer une connexion PostgreSQL.

Étape	Identifiants à utiliser	Pourquoi
Configuration de la source (cette page)	Un administrateur PostgreSQL, un superutilisateur ou le propriétaire de la table, connecté directement à la base de données source (par exemple, via psql ou la console de gestion de votre cloud provider).	La création de l'utilisateur de réplication, l'octroi de privilèges et la création de publications nécessitent des privilèges de superutilisateur ou de propriétaire de table que l'utilisateur de réplication ne possède pas. L'emplacement de réplication est créé par l'utilisateur de réplication lui-même, de sorte qu'un administrateur connecté à la base de données passe à ce rôle pour le créer. Pour la liste complète des privilèges, consultez les exigences relatives aux utilisateurs de bases de données PostgreSQL.
Pipeline de connexion et d'ingestion	L'utilisateur de réplication dédié (par exemple, databricks_replication) que vous créez lors de la configuration de la source.	La passerelle d'ingestion s'authentifie auprès de PostgreSQL en tant qu'utilisateur de réplication pour lire les modifications. Vous saisissez ces informations d'identification lorsque vous créez la connexion Unity Catalog. Voir Créer une connexion PostgreSQL.

remarque

Vous effectuez les tâches de configuration de la source en tant qu'administrateur, mais le pipeline d'ingestion n'utilise pas les identifiants d'administrateur. Seuls les identifiants de l'utilisateur de réplication sont stockés dans la connexion Unity Catalog.

Réplication logique pour la capture de données modifiées

Le connecteur PostgreSQL utilise la réplication logique pour suivre les changements dans les tables source. La réplication logique permet au connecteur de capturer les modifications de données (insertions, mises à jour et suppressions) sans nécessiter de Trigger ou de charge significative sur la base de données source.

La réplication logique Lakeflow PostgreSQL requiert les éléments suivants :

1. Lakeflow Connect prend en charge la réplication des données à partir de PostgreSQL version 13 et ultérieure.

2. Configurez la base de données pour la réplication logique :

   Pour AWS RDS et Aurora, définissez le paramètre rds.logical_replication sur 1.

3. Créez des publications qui incluent toutes les tables que vous souhaitez répliquer.

4. Créez des emplacements de réplication pour chaque catalogue qui sera répliqué.

remarque

Les publications doivent être créées avant de créer des emplacements de réplication.

Pour plus d'informations sur la réplication logique, consultez la documentation sur la réplication logique sur le site web de PostgreSQL.

Vue d’ensemble des tâches de configuration de la source

Effectuez les tâches suivantes dans PostgreSQL avant d'ingérer les données dans Databricks :

1. Vérifier PostgreSQL 13 ou une version ultérieure

2. Configurer l'accès au réseau (groupes de sécurité, règles de pare-feu ou VPN)

3. Configurer la réplication logique :

   • Pour AWS RDS/Aurora, définissez rds.logical_replication = 1.

   • Créez un utilisateur de réplication avec les privilèges requis. Consultez les exigences relatives aux utilisateurs de base de données PostgreSQL.

   • Définir l'identité des réplicas pour les tables. Voir Définir l'identité des réplicas pour les tables

   • Création de publications et d'emplacements de réplication

4. Facultatif : Configurez le suivi DDL en ligne pour la détection automatique des modifications de schéma. Si vous souhaitez opter pour le suivi DDL en ligne, contactez le support Databricks.

important

Si vous prévoyez de répliquer à partir de plusieurs bases de données PostgreSQL, vous devez créer une publication et un emplacement de réplication distincts pour chaque base de données. Le script de suivi DDL en ligne (s'il est utilisé) doit également être exécuté dans chaque base de données.

Configurer la réplication logique

Pour activer la réplication logique dans PostgreSQL, configurez les paramètres de la base de données et mettez en place les objets nécessaires.

Définissez le niveau WAL sur logique

Le Write-Ahead Log (WAL) doit être configuré pour la réplication logique. Ce paramètre nécessite généralement un redémarrage de la base de données.

1. Vérifiez le paramètre wal_level actuel :

   SQL

   SHOW wal_level;

2. Si la valeur n'est pas logical, définissez wal_level = logical dans la configuration du serveur et redémarrez le service PostgreSQL.

Créer un utilisateur de réplication

Créez un utilisateur PostgreSQL dédié pour l'ingestion Databricks avec des privilèges de réplication :

SQL

CREATE USER databricks_replication WITH PASSWORD 'your_secure_password';
GRANT CONNECT ON DATABASE your_database TO databricks_replication;
GRANT USAGE ON SCHEMA schema_name TO databricks_replication;
GRANT SELECT ON TABLE schema_name.table_name TO databricks_replication;
ALTER USER databricks_replication WITH REPLICATION;

Pour connaître les exigences détaillées en matière de privilèges, consultez les exigences relatives aux utilisateurs de base de données PostgreSQL.

Définir l’identité des réplicas pour les tables

Pour chaque table que vous souhaitez répliquer, configurez l'identité du réplica. Le paramètre correct dépend de la structure de la table :

Structure de la table	IDENTITÉ DE RÉPLICA Requise	Commande
La table dispose d'une clé primaire et ne contient pas de colonnes TOASTable (par exemple, TEXT, BYTEA, VARCHAR(n) avec des valeurs importantes)	DEFAULT	SQL ALTER TABLE schema_name.table_name REPLICA IDENTITY DEFAULT;
La table possède une clé primaire, mais inclut de grandes colonnes de longueur variable (TOASTable).	FULL	SQL ALTER TABLE schema_name.table_name REPLICA IDENTITY FULL;
La table n'a pas de clé primaire	FULL	SQL ALTER TABLE schema_name.table_name REPLICA IDENTITY FULL;

Structure de la table	IDENTITÉ DE RÉPLICA Requise	Commande
La table dispose d'une clé primaire et ne contient pas de colonnes TOASTable (par exemple, TEXT, BYTEA, VARCHAR(n) avec des valeurs importantes)	DEFAULT	SQL ALTER TABLE schema_name.table_name REPLICA IDENTITY DEFAULT;
La table possède une clé primaire, mais inclut de grandes colonnes de longueur variable (TOASTable).	FULL	SQL ALTER TABLE schema_name.table_name REPLICA IDENTITY FULL;
La table n'a pas de clé primaire	FULL	SQL ALTER TABLE schema_name.table_name REPLICA IDENTITY FULL;

Pour plus d'information sur les paramètres d'identité de réplication, consultez Replica Identity dans la documentation PostgreSQL.

Créer une publication

Créez une publication dans chaque base de données qui inclut les tables que vous souhaitez répliquer. Exécutez cette commande en tant que propriétaire de la table ou super-utilisateur :

SQL

-- Create a publication for specific tables
CREATE PUBLICATION databricks_publication FOR TABLE schema_name.table1, schema_name.table2;

-- Or create a publication for all tables in a database
CREATE PUBLICATION databricks_publication FOR ALL TABLES;

remarque

• Vous devez créer une publication distincte dans chaque base de données PostgreSQL que vous souhaitez répliquer.
• CREATE PUBLICATION ... FOR TABLE nécessite la propriété des tables listées. FOR ALL TABLES nécessite des privilèges de superutilisateur. Exécutez cette commande en tant que propriétaire de la table ou superutilisateur de la base de données, pas en tant qu'utilisateur de la réplication.
• Évitez d'ajouter des tables à la publication qui ne sont pas nécessaires à la réplication afin de réduire le trafic réseau inutile.

Configurer les paramètres de l'emplacement de réplication

Avant de créer des emplacements de réplication, configurez les paramètres de serveur suivants :

Limiter la rétention WAL pour les emplacements de réplication

parameter : max_slot_wal_keep_size

Il est **recommandé de ne pas définir** max_slot_wal_keep_size sur -1 (la valeur par default), car cela permet un gonflement illimité du WAL en raison de la rétention par des emplacements de réplication en retard ou inactifs. Selon votre charge de travail, définissez ce paramètre sur une valeur finie.

Pour en savoir plus sur le paramètre max_slot_wal_keep_size, consultez la documentation officielle de PostgreSQL.

remarque

Certains fournisseurs de cloud gérés n'autorisent pas la modification de ce parameter et s'appuient plutôt sur le monitoring intégré des slots et le nettoyage automatique. Examinez le comportement de la plateforme avant de configurer les alertes opérationnelles.

Pour plus d'informations, voir :

• Maîtriser la croissance des fichiers WAL PostgreSQL — pgDash
• PostgreSQL 13 : Ne laissez pas les slots tuer votre primaire | EDB

Configurer la capacité des emplacements de réplication

parameter : max_replication_slots

Chaque base de données PostgreSQL répliquée nécessite un emplacement de réplication logique. Définissez ce parameter à au moins le nombre de bases de données répliquées, ainsi que tout besoin de réplication existant.

Configurer les expéditeurs WAL

parameter : max_wal_senders

Ce paramètre définit le nombre maximal de processus émetteurs WAL concurrents qui transmettent des données WAL aux abonnés. Dans la plupart des cas, vous devriez disposer d'un processus émetteur WAL par emplacement de réplication pour garantir une réplication des données efficace et cohérente.

Configurez max_wal_senders pour qu'il soit au moins égal au nombre d'emplacements de réplication utilisés, en tenant compte de toute autre utilisation existante. Il est recommandé de le définir légèrement plus élevé afin d'offrir une flexibilité opérationnelle.

Créer un emplacement de réplication

Créez un emplacement de réplication dans chaque base de données que la passerelle d'ingestion Databricks utilisera pour suivre les changements. L'emplacement de réplication doit être créé par un utilisateur disposant du privilège REPLICATION. Si vous êtes connecté en tant que super-utilisateur ou administrateur, passez d'abord à l'utilisateur de réplication :

SQL

SET ROLE databricks_replication;

-- Databricks supports only the pgoutput plugin for replication slots
SELECT pg_create_logical_replication_slot('databricks_slot', 'pgoutput');

-- Switch back to the admin or table owner role for subsequent steps
RESET ROLE;

important

• Les emplacements de réplication conservent les données WAL jusqu'à ce qu'elles soient consommées par le connecteur. Configurez le paramètre max_slot_wal_keep_size pour limiter la rétention WAL et prévenir la croissance illimitée du WAL. Pour plus de détails, consultez Configurer les paramètres de l'emplacement de réplication.
• Lorsque vous supprimez un pipeline d'ingestion, vous devez supprimer manuellement l'emplacement de réplication associé. Voir Nettoyer les emplacements de réplication.

Facultatif : Configurer le suivi DDL intégré

Le suivi DDL intégré est une fonctionnalité facultative qui permet au connecteur de détecter et d'appliquer automatiquement les modifications de schéma de la base de données source. Cette fonctionnalité est désactivée par default.

attention

Le suivi DDL en ligne est actuellement en préversion et nécessite de contacter le support Databricks pour l'activer pour votre workspace.

Pour obtenir des information sur les modifications de schéma qui sont traitées automatiquement et celles qui nécessitent un refresh complet, consultez Comment les connecteurs gérés gèrent-ils l'évolution des schémas ? and évolution des schémas.

Mettre en place le suivi DDL en ligne

Si le suivi DDL en ligne a été activé pour votre Workspace, suivez ces étapes **dans chaque base de données PostgreSQL** :

1. download la dernière version du script :

download lakeflow_pg_ddl_change_tracking.sql

2. Exécuter le script :

   SQL

   \i lakeflow_pg_ddl_change_tracking.sql

   Le script crée les objets suivants dans le schéma public. Les noms d'objets incluent un suffixe de version (actuellement _1_0) qui suit la version du script :

   • Table d'audit : public.lakeflow_ddl_audit_table_1_0 — stocke les événements DDL capturés.
   • Fonctions de Trigger d'événements : public.lakeflow_ddl_audit_function_1_0 (pour ALTER TABLE événements) et public.lakeflow_drop_ddl_audit_function_1_0 (pour DROP TABLE événements).
   • Triggers d'événement : lakeflow_ddl_audit_trigger_1_0 (se déclenche le ddl_command_end) et lakeflow_drop_ddl_audit_trigger_1_0 (se déclenche le sql_drop).

3. Vérifiez que les Trigger et la table d’audit ont été créés avec succès :

   SQL

   -- Check for the DDL audit table
   SELECT * FROM pg_tables WHERE tablename LIKE 'lakeflow_ddl_audit_table%';
   
   -- Check for the event triggers
   SELECT * FROM pg_event_trigger WHERE evtname LIKE 'lakeflow%';

   Vous devriez voir la table d’audit lakeflow_ddl_audit_table_1_0 et deux event Trigger (lakeflow_ddl_audit_trigger_1_0 et lakeflow_drop_ddl_audit_trigger_1_0).

4. Ajoutez la table d'audit DDL à votre publication. Cette commande doit être exécutée en tant que propriétaire de la publication, et non en tant qu'utilisateur de la réplication :

   SQL

   ALTER PUBLICATION databricks_publication ADD TABLE public.lakeflow_ddl_audit_table_1_0;

Notes de configuration spécifiques au cloud

AWS RDS et Aurora

• Assurez-vous que le parameter rds.logical_replication est défini sur 1 dans le groupe de parameters.

• Configurez les groupes de sécurité pour autoriser les connexions depuis le Workspace Databricks.

• L'utilisateur de réplication nécessite le rôle rds_replication :

  SQL

  GRANT rds_replication TO databricks_replication;

Base de données Azure pour PostgreSQL

• Activez la réplication logique dans les paramètres du serveur via le portail Azure ou l'interface CLI.
• Configurez les règles de pare-feu pour autoriser les connexions depuis le workspace Databricks.
• Pour Flexible Server, la réplication logique est prise en charge. Pour Single Server, assurez-vous que vous utilisez un niveau pris en charge.

GCP Cloud SQL pour PostgreSQL

• Activez l'indicateur cloudsql.logical_decoding dans les paramètres de l'instance.
• Configurez les réseaux autorisés pour autoriser les connexions depuis le Databricks Workspace.
• Assurez-vous que l'indicateur cloudsql.enable_pglogical est défini sur on si vous utilisez des extensions pglogical.

Vérifier la configuration

Après avoir terminé les tâches de configuration, vérifiez que la réplication logique est correctement configurée :

1. Vérifiez que le wal_level est défini sur logical:

   SQL

   SHOW wal_level;

2. Vérifiez que l'utilisateur de réplication dispose du privilège replication :

   SQL

   SELECT rolname, rolreplication FROM pg_roles WHERE rolname = 'databricks_replication';

3. Vérifiez que l'utilisateur de réplication dispose des privilèges SELECT sur vos tables. Remplacez schema_name.table_name par le schéma et la table que vous répliquez (par exemple, public.my_table) :

   SQL

   SELECT has_table_privilege('databricks_replication', 'schema_name.table_name', 'SELECT');

4. Confirmez que la publication existe :

   SQL

   SELECT * FROM pg_publication WHERE pubname = 'databricks_publication';

5. Vérifier que l'emplacement de réplication existe :

   SQL

   SELECT slot_name, slot_type, active, restart_lsn
   FROM pg_replication_slots
   WHERE slot_name = 'databricks_slot';

6. Vérifiez l’identité des répliques pour vos tables :

   SQL

   SELECT schemaname, tablename, relreplident
   FROM pg_tables t
   JOIN pg_class c ON t.tablename = c.relname
   WHERE schemaname = 'your_schema';

   La colonne relreplident doit afficher d pour l'identité de réplica par default (utilise la clé primaire) ou f pour l'identité de réplica COMPLÈTE (requise pour les tables sans clé primaire ou avec des colonnes TOASTable).

Étapes suivantes

Après avoir terminé la configuration de la source, vous pouvez créer une passerelle et un pipeline d'ingestion pour ingérer des données depuis PostgreSQL. Consultez Ingérer des données depuis PostgreSQL.