Créer un connecteur source Cloud SQL pour PostgreSQL

Ce document explique comment créer un connecteur source Cloud SQL pour PostgreSQL pour Kafka Connect.

Un connecteur source Cloud SQL pour PostgreSQL est une instance d'un connecteur Debezium PostgreSQL. Il lit les modifications au niveau des lignes d'une base de données Cloud SQL pour PostgreSQL et les écrit dans des sujets d'un cluster Managed Service pour Apache Kafka.

Voici quelques cas d'utilisation de ce connecteur :

  • Surveillez les modifications apportées aux bases de données au niveau des lignes en temps réel.
  • Intégrez les événements de modification de la base de données dans une architecture événementielle.
  • Répondre aux événements de base de données tels que les insertions ou les suppressions de lignes.
  • Copiez les modifications apportées à la base de données dans d'autres systèmes.

Avant de commencer

Avant de créer un connecteur source Cloud SQL pour PostgreSQL, assurez-vous de disposer des éléments suivants :

Rôles et autorisations nécessaires

Pour obtenir les autorisations nécessaires pour créer un connecteur, demandez à votre administrateur de vous accorder le rôle IAM Éditeur de connecteurs Kafka gérés (roles/managedkafka.connectorEditor) sur votre projet. Pour en savoir plus sur l'attribution de rôles, consultez Gérer l'accès aux projets, aux dossiers et aux organisations.

Ce rôle prédéfini contient les autorisations requises pour créer un connecteur. Pour connaître les autorisations exactes requises, développez la section Autorisations requises :

Autorisations requises

Les autorisations suivantes sont requises pour créer un connecteur :

  • Créez un connecteur : managedkafka.connectors.create

Vous pouvez également obtenir ces autorisations avec des rôles personnalisés ou d'autres rôles prédéfinis.

Accorder des autorisations de lecture à partir de Cloud SQL

Le compte de service Managed Kafka doit être autorisé à accéder à Cloud SQL pour PostgreSQL. Attribuez les rôles IAM suivants au compte de service :

  • Client Cloud SQL (roles/cloudsql.client)
  • Utilisateur d'instance Cloud SQL (roles/cloudsql.instanceUser)

Le compte de service Managed Kafka a le format suivant : service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com, où PROJECT_NUMBER est le numéro de projet du cluster Connect.

Si votre cluster Connect se trouve dans un projet différent de votre cluster Managed Service pour Apache Kafka, consultez Créer un cluster Connect dans un autre projet.

Configurer la base de données

Avant de créer le connecteur, vous devez configurer la réplication de la base de données et permettre au connecteur de s'authentifier auprès de la base de données. Les sections suivantes décrivent ces étapes.

Activer le décodage logique

Un connecteur source Cloud SQL pour PostgreSQL s'appuie sur la fonctionnalité de décodage logique de PostgreSQL. Pour activer le décodage logique dans votre instance Cloud SQL pour PostgreSQL, procédez comme suit.

Console

  1. Accédez à Cloud SQL > Instances.

    Accéder à la page "Instances"

  2. Cliquez sur le nom de l'instance.

  3. Cliquez sur Modifier.

  4. Développez Options et paramètres.

  5. Cliquez sur Ajouter un indicateur de base de données.

  6. Dans la liste Sélectionner un indicateur, sélectionnez cloudsql.logical_decoding.

  7. Pour Valeur, sélectionnez On.

  8. Cliquez sur OK.

  9. Cliquez sur Enregistrer.

Pour en savoir plus, consultez Configurer la réplication logique et le décodage.

Configurer la capture de données modifiées (CDC)

Après avoir activé le décodage logique dans votre instance, activez la capture de données modifiées (CDC) pour les tables que vous souhaitez répliquer.

Pour activer la CDC pour une table, exécutez l'instruction SQL CREATE PUBLICATION. Cette instruction crée une publication, qui définit un groupe de tables à répliquer.

  • Option 1. Créez une publication qui réplique les modifications apportées à toutes les tables de la base de données.

    CREATE PUBLICATION dbz_publication FOR ALL TABLES;

  • Option 2. Créez une publication pour un ensemble spécifique de tables.

    CREATE PUBLICATION dbz_publication FOR TABLE TABLE_LIST;

    Remplacez TABLE_LIST par une liste de tables séparées par une virgule, au format "schema_name"."table_name". En encadrant les noms de schémas et de tables de guillemets doubles, comme indiqué, vous évitez les erreurs de syntaxe si les noms contiennent des caractères spéciaux ou des majuscules.

Par défaut, le connecteur utilise dbz_publication comme nom de publication. Pour utiliser une publication portant un autre nom, consultez Nom de la publication.

Créer un compte utilisateur pour le compte de service Managed Kafka

Le connecteur de source Cloud SQL pour PostgreSQL utilise l'authentification IAM pour les bases de données afin de se connecter à la base de données. Pour activer l'authentification IAM pour les bases de données, ajoutez le compte de service Managed Kafka à l'instance Cloud SQL, comme suit :

Console

  1. Accédez à Cloud SQL > Instances.

    Accéder à la page "Instances"

  2. Cliquez sur le nom de l'instance.

  3. Dans le volet de navigation, cliquez sur Utilisateurs.

  4. Cliquez sur Ajouter un compte utilisateur.

  5. Dans le volet Ajouter un compte utilisateur, sélectionnez Cloud IAM.

  6. Dans le champ Compte principal IAM, saisissez ce qui suit :

    service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com

    Remplacez PROJECT_NUMBER par le numéro de projet du cluster Connect.

  7. Cliquez sur Ajouter.

gcloud

Exécutez la commande gcloud sql users create :

gcloud sql users create service-PROJECT_NUMBER@gcp-sa-managedkafka.iam \
  --instance=INSTANCE_NAME \
  --type=cloud_iam_service_account

Remplacez les éléments suivants :

  • PROJECT_NUMBER : numéro de projet du cluster Connect.

  • INSTANCE_NAME : nom de l'instance Cloud SQL pour PostgreSQL.

En raison de la limite de longueur d'un nom d'utilisateur de base de données, le suffixe .gserviceaccount.com est supprimé du nom d'utilisateur, qui devient donc service-PROJECT_NUMBER@gcp-sa-managedkafka.iam. Lorsque vous exécutez des requêtes SQL qui font référence au compte utilisateur IAM, spécifiez le nom tronqué.

Configurer le compte utilisateur

Après avoir créé le compte utilisateur IAM, connectez-vous à la base de données en tant qu'utilisateur disposant du rôle cloudsqlsuperuser (tel que l'utilisateur postgres par défaut) et exécutez les requêtes SQL suivantes.

Console

  1. Permet à l'utilisateur de lire le journal WAL.

    ALTER USER "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam" WITH REPLICATION;

  2. Accordez à l'utilisateur l'autorisation SELECT sur les tables.

    GRANT SELECT ON ALL TABLES IN SCHEMA "SCHEMA_NAME"
TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";

    Vous pouvez également accorder l'autorisation SELECT sur des tables individuelles. Si vous choisissez cette option, vous devez également définir la propriété de configuration table.include.list du connecteur sur la liste des tables autorisées. La requête SQL suivante accorde l'autorisation SELECT sur une seule table :

    GRANT SELECT ON TABLE SCHEMA_NAME.TABLE_NAME
TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";

  3. Pour chaque table, accordez à l'utilisateur l'accès au schéma de la table. Vous pouvez ignorer cette étape si la table se trouve dans le schéma public par défaut.

    GRANT USAGE ON SCHEMA SCHEMA_NAME
TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";

Configurer la mise en réseau

Un connecteur source Cloud SQL pour PostgreSQL peut se connecter à l'instance Cloud SQL de différentes manières :

  • Adresse IP privée
  • Private Service Connect
  • Adresse IP publique

Pour en savoir plus sur ces options, consultez Choisir comment se connecter à Cloud SQL. Pour respecter les bonnes pratiques en matière de sécurité, nous vous recommandons d'utiliser une adresse IP privée ou Private Service Connect, car ces options ne nécessitent pas de connexion à une adresse IP externe.

Le tableau suivant présente les exigences réseau pour chaque option :

Type d'adresse IP Conditions requises
Adresse IP privée Configurez une adresse IP privée pour votre instance. Pour en savoir plus, consultez Configurer une adresse IP privée.
Private Service Connect
  1. Configurez Private Service Connect pour votre instance et obtenez le nom DNS du point de terminaison Private Service Connect. Pour en savoir plus, consultez Se connecter à une instance à l'aide de Private Service Connect.
  2. Ajoutez le nom DNS du point de terminaison à la section Domaines DNS pouvant être résolus du cluster Connect. Pour en savoir plus, consultez Mettre à jour un cluster Connect.
Adresse IP publique
  1. Configurez une adresse IP publique pour votre instance. Pour en savoir plus, consultez Configurer une adresse IP publique.
  2. Configurez Public NAT pour permettre aux nœuds de calcul du cluster Connect de communiquer avec Internet. Pour en savoir plus, consultez Configurer Public NAT. Lorsque vous créez la passerelle Cloud NAT, spécifiez le réseau VPC qui contient le sous-réseau principal du cluster Connect.

Créer un connecteur source Cloud SQL pour PostgreSQL

Pour créer un connecteur source Cloud SQL pour PostgreSQL, procédez comme suit.

Lorsque le connecteur est initialisé, il effectue les actions suivantes :

  1. Crée un instantané initial de la base de données.
  2. Crée un sujet Kafka pour chaque table comportant des lignes.
  3. Pour chaque ligne de la base de données, envoie un événement de modification au thème correspondant.

Pendant l'exécution du connecteur, il continue d'envoyer des événements de modification aux thèmes. Pour en savoir plus sur l'instantané initial, consultez Instantanés dans la documentation Debezium.

Console

  1. Dans la console Google Cloud , accédez à la page Connecter des clusters.

    Accéder à Connect Clusters

  2. Cliquez sur le cluster Connect dans lequel vous souhaitez créer le connecteur.

  3. Cliquez sur Créer un connecteur.

  4. Saisissez une chaîne pour le nom du connecteur.

    Pour obtenir des instructions sur la façon de nommer un connecteur, consultez les consignes de dénomination des ressources Managed Service pour Apache Kafka.

  5. Pour Plug-in de connecteur, sélectionnez Source Cloud SQL pour PostgreSQL.

  6. Dans la liste Instance, sélectionnez l'instance Cloud SQL.

  7. Dans la liste Base de données, sélectionnez la base de données Cloud SQL.

  8. Dans le champ Préfixe du sujet, saisissez un préfixe à utiliser pour les noms des sujets Kafka. Choisissez un préfixe unique pour chaque connecteur de source Cloud SQL pour PostgreSQL.

  9. Facultatif : Dans le champ Noms des tables, saisissez une liste de tables séparées par une virgule à partir desquelles lire les données de modification, au format "schema_name"."table_name". Si vous laissez ce champ vide, le connecteur lit les données de modification de toutes les tables non système de la base de données.

  10. Facultatif : Dans la zone Configurations, ajoutez des propriétés de configuration ou modifiez les propriétés par défaut. Pour en savoir plus, consultez Configurer le connecteur.

    Vous devrez peut-être remplacer les valeurs par défaut des propriétés suivantes :

  11. Facultatif : Sélectionnez la Règle de redémarrage des tâches. Pour en savoir plus, consultez la section Règles de redémarrage des tâches.

  12. Cliquez sur Créer.

gcloud

  1. Dans la console Google Cloud , activez Cloud Shell.

    Activer Cloud Shell

    En bas de la console Google Cloud , une session Cloud Shell démarre et affiche une invite de ligne de commande. Cloud Shell est un environnement shell dans lequel Google Cloud CLI est déjà installé, et dans lequel des valeurs sont déjà définies pour votre projet actuel. L'initialisation de la session peut prendre quelques secondes.

  2. Exécutez la commande gcloud managed-kafka connectors create :

    gcloud managed-kafka connectors create CONNECTOR_ID \
    --location=LOCATION \
    --connect-cluster=CONNECT_CLUSTER_ID \
    --config-file=CONFIG_FILE

    Remplacez les éléments suivants :

    • CONNECTOR_ID : ID ou nom du connecteur. Pour obtenir des instructions sur la façon de nommer un connecteur, consultez les consignes de dénomination des ressources Managed Service pour Apache Kafka. Le nom d'un connecteur est immuable.

    • LOCATION : emplacement dans lequel vous créez le connecteur. Il doit s'agir du même emplacement que celui où vous avez créé le cluster Connect.

    • CONNECT_CLUSTER_ID : ID du cluster Connect dans lequel le connecteur est créé.

    • CONFIG_FILE : chemin d'accès au fichier de configuration YAML du connecteur.

    Voici un exemple de fichier de configuration pour le connecteur source Cloud SQL pour PostgreSQL :

    connector.class: io.debezium.connector.postgresql.PostgresConnector
database.dbname: DATABASE_NAME
driver.cloudSqlInstance: INSTANCE_ID
driver.enableIamAuth: "true"
driver.ipTypes: IP_TYPES
driver.sslmode: disable
key.converter: org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable: "false"
plugin.name: pgoutput
slot.name: SLOT_NAME
table.include.list: TABLE_LIST
topic.prefix: TOPIC_PREFIX
value.converter: org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable: "true"

    Remplacez les éléments suivants :

    • INSTANCE_ID : ID de l'instance Cloud SQL contenant la base de données, au format suivant :

      PROJECT_ID:REGION:INSTANCE_NAME

    • DATABASE_NAME : nom de la base de données Cloud SQL à partir de laquelle lire les données.

    • IP_TYPES : liste de types d'adresses IP séparés par une virgule

    • SLOT_NAME : nom de l'emplacement de réplication à créer.

    • TABLE_LIST : liste de tables séparées par des virgules à partir desquelles lire les données de modification, au format "schema_name"."table_name".

    • TOPIC_PREFIX : préfixe à utiliser pour les noms de sujets Kafka.

Configurer le connecteur

Cette section décrit certaines propriétés de configuration que vous pouvez définir sur le connecteur. Pour obtenir la liste complète, consultez Connecteur Debezium pour PostgreSQL dans la documentation Debezium.

Types d'adresses IP

La propriété driver.ipTypes spécifie le type d'adresse IP que le connecteur utilise pour se connecter à la base de données :

  • PRIVATE : adresse IP privée
  • PSC : Private Service Connect
  • PUBLIC : adresse IP publique

La propriété driver.ipTypes contient une liste de types d'adresses IP séparés par une virgule, dans l'ordre de préférence. Par exemple, driver.ipTypes=PRIVATE,PUBLIC.

Pour en savoir plus, consultez Configurer la mise en réseau.

Titre de la publication

Par défaut, le connecteur tente de diffuser des données à partir d'une publication nommée dbz_publication. Pour spécifier une autre publication, ajoutez publication.name=PUBLICATION_NAME à la configuration, où PUBLICATION_NAME correspond au nom de la publication. Exemple : publication.name=my_publication.

Emplacements de réplication

PostgreSQL utilise des emplacements de réplication pour diffuser en flux continu les modifications apportées aux tables de la base de données. Par défaut, le connecteur crée un emplacement de réplication nommé debezium. Pour utiliser un autre nom d'emplacement, définissez la propriété slot.name.

Si vous créez deux instances du connecteur pour la même base de données, vous devez spécifier un nom d'emplacement unique pour chaque connecteur.

Par défaut, le connecteur définit la propriété slot.drop.on.stop sur false pour éviter toute perte de données. Lorsque vous supprimez définitivement un connecteur, vous devez supprimer manuellement l'emplacement de réplication qu'il utilisait. Le nom de l'emplacement de réplication est défini par défaut sur debezium, sauf s'il est configuré différemment à l'aide de la propriété slot.name.

Nous vous recommandons de configurer des alertes pour surveiller l'utilisation du disque WAL sur votre serveur de base de données PostgreSQL source et de supprimer les emplacements de réplication inutilisés.

Filtre de tableau

Par défaut, le connecteur capture les données de modification de chaque table non système de la base de données. Pour filtrer les tables à capturer, spécifiez un ou plusieurs des paramètres suivants :

  • schema.include.list : liste des schémas à inclure.
  • schema.exclude.list : liste des schémas à exclure. Ne peut pas être utilisé avec schema.include.list.
  • table.include.list : liste des tables à inclure.
  • table.exclude.list : liste des tables à exclure. Ne peut pas être utilisé avec table.include.list.

Noms des thèmes

Par défaut, le connecteur crée des sujets Kafka en suivant la convention d'attribution de noms suivante : topic_prefix.schema.table_name, où topic.prefix correspond à la valeur de la configuration topic.prefix.

Pour en savoir plus, consultez Noms de sujets dans la documentation Debezium.

Étapes suivantes