Ajouter la CDC de la base de données PostgreSQL en tant que source dans le hub en temps réel

Cet article explique comment ajouter la capture des changements de données (CDC) de la base de données PostgreSQL en tant que source d’événements dans le hub En temps réel Fabric.

Le connecteur source PostgreSQL de Capture de Données Modifiées (CDC) pour les flux d'événements de Microsoft Fabric permet de capturer un instantané des données au moment présent dans une base de données PostgreSQL. Actuellement, postgreSQL Database Change Data Capture (CDC) est pris en charge à partir des services suivants où les bases de données sont accessibles publiquement :

• Base de données Azure pour PostgreSQL
• Amazon RDS pour PostgreSQL
• Amazon Aurora PostgreSQL
• Google Cloud SQL pour PostgreSQL

Une fois la source CDC de base de données PostgreSQL ajoutée au flux d’événements, elle capture les modifications au niveau des lignes apportées aux tables spécifiées. Ces modifications peuvent ensuite être traitées en temps réel et envoyées à différentes destinations pour une analyse plus approfondie.

Prérequis

• Accès à un espace de travail en mode de licence de capacité Fabric ou en mode de licence d’évaluation avec des autorisations de type Collaborateur ou supérieures.

• Accès utilisateur inscrit dans la base de données PostgreSQL.

• Votre base de données PostgreSQL doit être accessible publiquement et ne doit pas se trouver derrière un pare-feu ou sécurisé dans un réseau virtuel. S’il réside dans un réseau protégé, connectez-vous à celui-ci à l’aide de l’injection de réseau virtuel du connecteur Eventstream.

• CDC activé dans la base de données et les tables PostgreSQL.

  Si vous avez Azure Database pour PostgreSQL, suivez les étapes de la section suivante pour activer la CDC. Pour des informations détaillées, consultez Réplication logique et décodage logique - Serveur flexible Azure Database pour PostgreSQL.

  Pour les autres bases de données PostgreSQL, consultez le connecteur Debezium pour PostgreSQL :: Documentation de Debezium.

• Si vous n’avez pas de flux d’événements, créez un flux d’événements.

Activer la capture de données modifiées dans votre base de données PostgreSQL

Cette section utilise Azure Database pour PostgreSQL comme exemple.

Procédez comme suit pour activer CDC sur votre serveur flexible Azure Database pour PostgreSQL :

1. Sur la page de votre Azure Database pour PostgreSQL Flexible Server dans le portail Azure, sélectionnez Paramètres du serveur dans le menu de navigation.

2. Sur la page Paramètres du serveur :

   • Définir wal_level à logical.
   • Mettez à jour le max_worker_processes sur au moins 16.

   

3. Enregistrez les modifications et redémarrez le serveur.

4. Vérifiez que votre instance de serveur flexible Azure Database pour PostgreSQL autorise le trafic réseau public.

5. Accordez les autorisations de réplication de l’utilisateur administrateur en exécutant l’instruction SQL suivante. Si vous souhaitez utiliser un autre compte utilisateur pour connecter votre base de données PostgreSQL (DB) afin de récupérer les données de changement (CDC), vérifiez que l’utilisateur est le propriétaire de la table.

   ALTER ROLE <admin_user_or_table_owner_user> WITH REPLICATION;
   

Accéder à la page Sources de données

1. Connectez-vous à Microsoft Fabric.

2. Si vous voyez Power BI en bas à gauche de la page, basculez vers la charge de travail Fabric en sélectionnant Power BI, puis en sélectionnant Fabric.

   

3. Sélectionnez En temps réel dans la barre de navigation gauche.

   

4. La page de données de streaming s’ouvre par défaut. Cliquez sur le bouton Ajouter des données pour accéder à la page Sources de données .

   

   Vous pouvez également accéder directement à la page Sources de données en sélectionnant l’option Ajouter des données dans la barre de navigation de gauche.

   

Sélectionner la base de données PostgreSQL CDC comme type de source

Sur la page Sources de données, sélectionnez la catégorie Sources Microsoft en haut, puis sélectionnez Se connecter dans la vignette Azure DB pour PostgreSQL (CDC).

Configurer la source CDC d'Azure Database pour PostgreSQL

Ingérer des données modifiées à partir de bases de données PostgreSQL avec inscription automatique de schéma de table via CDC dans Eventstream.

1. Sur la page Connecter, sélectionnez Nouvelle connexion.

   

2. Dans la section Paramètres de connexion, saisissez les informations suivantes :

   • Serveur : adresse du serveur de votre base de données PostgreSQL, par exemple my-pgsql-server.postgres.database.azure.com.

   • Base de données : nom de la base de données, par exemple my_database.

     

   • Nom de la connexion : saisissez un nom pour la connexion.

   • Type d’authentification, sélectionnez Basic et entrez votre nom d’utilisateur et votre mot de passe pour la base de données.

     Remarque

     Actuellement, les flux d’événements Fabric prennent uniquement en charge l’authentification de base .

   • Sélectionnez Se connecter pour terminer les paramètres de connexion.

3. Port : entrez le numéro de port de votre serveur. La valeur par défaut est 5432. Si votre connexion cloud sélectionnée est configurée dans Gérer les connexions et les passerelles, vérifiez que le numéro de port correspond à celui défini ici. S’ils ne correspondent pas, le numéro de port dans la connexion cloud dans Gérer les connexions et les passerelles est prioritaire.

4. Vous pouvez choisir entre deux options lors de la capture des modifications à partir de tables de base de données :

   • Toutes les tables : capturez les modifications de chaque table de la base de données.
   • Entrez le ou les noms de table : vous permet de spécifier un sous-ensemble de tables à l’aide d’une liste séparée par des virgules. Vous pouvez utiliser soit : identificateurs de table complets au format schemaName.tableName ou expressions régulières valides. Exemples :
   • dbo.test.*: sélectionnez toutes les tables dont les noms commencent par test dans le schéma dbo.
   • dbo\.(test1|test2): Sélectionner dbo.test1 et dbo.test2.

   Vous pouvez combiner les deux formats dans la liste. La limite de caractères totale de l’entrée entière est de 102 400 caractères.

5. Nom de l’emplacement (facultatif) : entrez le nom de l’emplacement de décodage logique PostgreSQL créé pour la diffusion en continu des modifications à partir d’un plug-in particulier pour une base de données/un schéma particulier. Le serveur utilise cet emplacement pour diffuser des événements vers le connecteur de streaming Eventstream. Il doit contenir uniquement des lettres minuscules, des chiffres et des traits de soulignement.

   • S’il n’est pas spécifié, un GUID est utilisé pour créer l’emplacement, nécessitant les autorisations de base de données appropriées.
   • Si un nom d’emplacement spécifié existe, le connecteur l’utilise directement.

6. Développez les paramètres avancés pour accéder à des options de configuration supplémentaires pour la source CDC de la base de données PostgreSQL :

   • Nom de la publication : spécifie le nom de la publication de réplication logique PostgreSQL à utiliser. Cette valeur doit correspondre à une publication existante dans la base de données, ou elle est automatiquement créée en fonction du mode de création automatique. Valeur par défaut : dbz_publication.

     Remarque

     L’utilisateur du connecteur doit disposer des autorisations de superutilisateur pour créer la publication. Nous vous recommandons de créer la publication manuellement avant de démarrer le connecteur pour la première fois afin d’éviter les problèmes liés aux autorisations.

   • Mode de création automatique de publication : contrôle si et comment la publication est créée automatiquement. Les options sont les suivantes :

     • Filtered (valeur par défaut) : si la publication spécifiée n'existe pas, le connecteur en crée une qui inclut uniquement les tables sélectionnées (comme spécifié dans la liste des tables à inclure).
     • AllTables: si la publication spécifiée existe, le connecteur l’utilise. S’il n’existe pas, le connecteur en crée un qui inclut toutes les tables de la base de données.
     • Disabled: le connecteur ne crée pas de publication. Si la publication spécifiée est manquante, le connecteur lève une exception et s’arrête. Dans ce cas, la publication doit être créée manuellement dans la base de données.

     Pour plus d’informations, consultez la documentation Debezium sur le mode de création automatique de publication

   • Mode de gestion décimal : spécifie la façon dont le connecteur gère les valeurs postgreSQL DECIMAL et NUMERIC de colonne :

     • Precise: représente des valeurs utilisant des types décimaux exacts (par exemple, Java BigDecimal) pour garantir une précision et une précision complètes dans la représentation des données.
     • Double: convertit les valeurs en nombres à virgule flottante de double précision. Cette option améliore la facilité d’utilisation et les performances, mais peut entraîner une perte de précision.
     • String: encode les valeurs sous forme de chaînes mises en forme. Cette option facilite leur consommation dans les systèmes en aval, mais perd des informations sémantiques sur le type numérique d’origine.

   • Mode instantané : spécifiez les critères d’exécution d’un instantané au démarrage du connecteur :

     • Initial: Le connecteur exécute un instantané uniquement lorsqu'aucun décalage n’a été enregistré pour le nom du serveur logique, ou s’il détecte qu’un instantané antérieur n’a pas été achevé avec succès. Une fois l’instantané terminé, le connecteur commence à diffuser en continu les enregistrements d’événements pour les modifications de base de données suivantes.
     • InitialOnly: Le connecteur effectue un instantané uniquement lorsque aucun décalage n’a été enregistré pour le nom du serveur logique. Une fois que l’instantané est terminé, le connecteur s’arrête. Il ne passe pas en mode streaming pour lire les événements de modification du binlog.
     • NoData: le connecteur exécute un instantané qui capture uniquement le schéma, mais pas les données de table. Définissez cette option si vous n'avez pas besoin d'un instantané cohérent des données, mais que vous avez uniquement besoin des modifications depuis le démarrage du connecteur.

   • Requête de pulsation : définit une requête que le connecteur exécute sur la base de données source lors de l'envoi d'un message de pulsation par le connecteur.

   • Surcharge de l’instruction SELECT pour l’instantané : spécifie les lignes de la table à inclure dans un instantané. Utilisez la propriété si vous souhaitez qu’un instantané inclue uniquement un sous-ensemble des lignes d’une table. Cette propriété affecte uniquement les instantanés. Cela ne s’applique pas aux événements que le connecteur lit dans le journal.

Flux ou détails de la source

1. Dans la page Se connecter , suivez l’une de ces étapes en fonction de l’utilisation d’Eventstream ou de Real-Time hub.

   • Flux d’événements :

     Dans le volet Détails de la source à droite, procédez comme suit :

     1. Pour le nom de la source, sélectionnez l'icône du crayon pour modifier le nom.

     2. Notez que le nom eventstream et le nom stream sont en lecture seule.

   • Hub en temps réel :

     Dans la section Détails du flux à droite, procédez comme suit :

     1. Sélectionnez l’espace de travail Fabric dans lequel vous souhaitez créer le flux d’événements.

     2. Pour le flux d'événements, sélectionnez le bouton Crayon et entrez un nom pour le flux d'événements.

     3. La valeur du nom du flux est générée automatiquement pour vous en ajoutant -stream au nom de l’événementstream. Ce flux s’affiche sur la page Tous les flux de données du hub en temps réel une fois que l’Assistant est terminé.

2. Sélectionnez Suivant en bas de la page Configurer .

Vérifier et se connecter

Dans l’écran Vérifier + se connecter , passez en revue le résumé, puis sélectionnez Ajouter (Eventstream) ou Se connecter (Real-Time hub).

Page Gestion des schémas

1. Dans l’étape de gestion des schémas , choisissez l’une des options suivantes :

   • Événements prêts à l’analytique et schéma mis à jour automatiquement (préversion DeltaFlow) : le connecteur transforme les événements CDC bruts en flux prêts pour l’analytique qui reflètent la structure de votre table source. DeltaFlow enrichit les événements avec des métadonnées telles que le type de modification (insertion, mise à jour ou suppression) et les horodatages, et gère automatiquement les tables de destination et l’évolution du schéma.
   • Événements CDC bruts : le connecteur ingère et rend disponibles les événements CDC bruts. Si vous le souhaitez, le connecteur peut découvrir automatiquement des schémas de table et les inscrire dans le registre de schémas. Utilisez cette option lorsque vous souhaitez connaître le schéma sans transformation DeltaFlow.

   Remarque

   La capture d’écran suivante montre Azure SQL Database CDC. Les options de gestion des schémas sont identiques pour tous les connecteurs sources CDC pris en charge.

   

2. Activer l’association de schéma d’événement.

3. Pour Workspace, sélectionnez un espace de travail Fabric pour le jeu de schémas.

4. Pour le jeu de schémas, + Créer est sélectionné par défaut, ce qui crée un jeu de schémas. Vous pouvez le modifier pour sélectionner un jeu de schémas d’événements existant.

5. Si vous avez sélectionné l’option + Créer à l’étape précédente, entrez un nom pour le jeu de schémas.

6. Dans la page Vérifier + se connecter , passez en revue le résumé, puis sélectionnez Ajouter (Eventstream) ou Se connecter (Real-Time hub).

   

   Pour toutes les tables ou tables sélectionnées dans la base de données PostgreSQL, le connecteur effectue une découverte automatique et crée des schémas et les inscrit auprès du registre de schémas.

DeltaFlow : transformation d'événements prête pour l'analyse (aperçu)

Lorsque vous activez les événements prêts pour l’analytique et le schéma mis à jour automatiquement (DeltaFlow), le connecteur fournit les fonctionnalités suivantes :

• Forme d’événement prête pour l’analytique : les événements bruts de capture de données modifiées de Debezium sont transformés en un format tabulaire qui reflète la structure de la table source. Les événements sont enrichis avec des colonnes de métadonnées, notamment le type de modification (insert, updateou delete) et l’horodatage des événements.
• Gestion automatique des tables de destination : lorsque vous routez des flux deltaFlow vers une destination prise en charge comme un eventhouse, les tables de destination sont automatiquement créées pour correspondre au schéma de table source. Vous n’avez pas besoin de créer ou de configurer manuellement des tables de destination.
• Gestion de l’évolution du schéma : lorsque les tables de base de données sources changent (par exemple, les nouvelles colonnes sont ajoutées ou des tables sont créées), DeltaFlow détecte automatiquement les modifications, met à jour les schémas inscrits et ajuste les tables de destination en conséquence. Ce comportement réduit l’intervention manuelle causée par les modifications de schéma.

Pour plus d’informations sur la façon dont DeltaFlow transforme les événements CDC bruts en sortie prête pour l’analytique, y compris les types d’opérations et les colonnes de métadonnées, consultez la transformation de sortie DeltaFlow.

Afficher les détails du flux de données

1. Dans la page Vérifier + se connecter, si vous sélectionnez Ouvrir un eventstream, l’assistant ouvre l’eventstream qu’il a créé pour vous avec la CDC PostgreSQL Database sélectionné comme source. Pour fermer l’assistant, sélectionnez Terminer en bas de la page.

   

2. Vous devez voir le flux dans la section Données de streaming récentes de la page d’accueil du hubReal-Time . Pour obtenir des étapes détaillées, consultez Afficher les détails des flux de données dans le hub En temps réel de Fabric.

Contenu connexe

Pour en savoir plus sur l’utilisation de flux de données, consultez les articles suivants :

• Traitement des flux de données
• Analyse des flux de données
• Définir des alertes sur des flux de données