Data Transmission Service:Consommer les données d'abonnement avec Kafka

Notes

• Lorsque vous utilisez la démonstration présentée dans cette rubrique, la validation automatique peut entraîner une perte de données, car les décalages (offsets) peuvent être validés avant la fin du traitement des données. Pour éviter ce problème, nous recommandons d'utiliser la validation manuelle.

  Remarque

  En cas d'échec de la validation, le client redémarre la consommation à partir du dernier point de contrôle enregistré. Cela peut provoquer une duplication des données, que vous devez gérer dans la logique de votre application.

• Les données sont stockées via la sérialisation Avro. Pour plus de détails sur le format, consultez le fichier Record.avsc.

  Avertissement

  Si vous utilisez un client Kafka autre que celui fourni dans cette rubrique, les données risquent d'être incorrectement analysées lors de la désérialisation (Exemple de désérialisation Avro DTS). Vous devez vérifier l'intégrité des données.

• Dans DTS, l'API offsetForTimes utilise la seconde comme unité de temps, tandis que l'API native Kafka utilise la milliseconde.

• Le serveur d'abonnement aux données peut subir des déconnexions réseau transitoires dues à des événements tels qu'une reprise après sinistre. Si vous n'utilisez pas le client Kafka fourni dans cette rubrique, assurez-vous que votre client prend en charge les nouvelles tentatives de connexion réseau.

• Si vous utilisez un client Kafka natif en mode d'abonnement, le point de contrôle de consommation stocké sur le serveur peut être effacé lorsque DTS change son module de collecte de données incrémentielles. Dans ce cas, vous devez ajuster manuellement le point de contrôle de consommation. Si vous devez utiliser le mode d'abonnement, nous vous recommandons d'utiliser le SDK fourni par DTS pour consommer les données d'abonnement et gérer vous-même le point de contrôle de consommation. Pour plus d'informations, consultez les rubriques Consommer des données d'abonnement à l'aide du SDK et Gérer le point de contrôle de consommation.

Environnement d'exécution du client Kafka

Téléchargez la démonstration du client Kafka. Pour obtenir des instructions d'utilisation, consultez le fichier Readme de la démonstration.

Procédure

Exécutez la démonstration du client Kafka dans IntelliJ IDEA pour consommer les données d'un canal d'abonnement aux données.

1. Créez un canal d'abonnement aux données. Pour plus d'informations, consultez la rubrique Présentation des solutions d'abonnement.

2. Créez un ou plusieurs groupes de consommateurs. Pour plus d'informations, consultez la rubrique Créer un groupe de consommateurs.

3. Téléchargez la démonstration du client Kafka et décompressez-la.

   Remarque

   Cliquez sur , puis sélectionnez Download ZIP pour télécharger le fichier.

4. Ouvrez IntelliJ IDEA et cliquez sur Open.

5. Dans la boîte de dialogue, accédez au répertoire dans lequel vous avez décompressé la démonstration du client Kafka. Développez les dossiers comme illustré dans la figure suivante pour trouver le fichier Project Object Model (POM) : pom.xml.

6. Dans la boîte de dialogue qui s'affiche, cliquez sur Open as Project.

7. Dans la fenêtre IntelliJ IDEA, développez les dossiers, puis recherchez et double-cliquez sur le fichier de démonstration du client Kafka : NotifyDemoDB.java.

8. Définissez les valeurs des paramètres dans le fichier NotifyDemoDB.java.

   Paramètre	Description	Comment l'obtenir
   USER_NAME	Nom d'utilisateur du groupe de consommateurs. Avertissement Si vous utilisez un client différent, vous devez définir le nom d'utilisateur au format <compte du groupe de consommateurs>-<ID du groupe de consommateurs>. Exemple : dtstest-dtsaebpv.	Dans la console DTS, cliquez sur l'ID de l'instance d'abonnement cible. Dans le volet de navigation de gauche, cliquez sur Consume Data. Sur la page qui s'affiche, vous trouverez les informations relatives à l'Consumer Group ID/Name et au Account du groupe de consommateurs. Remarque Le mot de passe du compte du groupe de consommateurs est spécifié lors de la création du groupe de consommateurs.
   PASSWORD_NAME	Mot de passe du compte du groupe de consommateurs.
   SID_NAME	ID du groupe de consommateurs.
   GROUP_NAME	Cette valeur doit correspondre à l'ID du groupe de consommateurs.
   KAFKA_TOPIC	Sujet d'abonnement du canal d'abonnement aux données.	Dans la console DTS, cliquez sur l'ID de l'instance d'abonnement cible. Sur la page Basic Information, vous trouverez les informations relatives au Topic et au Network.
   KAFKA_BROKER_URL_NAME	Adresse réseau du canal d'abonnement aux données. Remarque Si votre client Kafka et votre canal d'abonnement aux données se trouvent dans le même réseau classique ou VPC, utilisez l'adresse IP interne pour minimiser la latence réseau. Nous déconseillons l'utilisation d'un endpoint public en raison d'une éventuelle instabilité du réseau.
   INITIAL_CHECKPOINT_NAME	Heure de début de la consommation des données, spécifiée sous forme d'horodatage UNIX. Exemple : 1592269238. Remarque L'enregistrement de l'horodatage vous permet de : Reprendre la consommation à partir du dernier horodatage enregistré après une interruption pour éviter toute perte de données. Commencer à consommer les données à partir d'un moment précis. Si SUBSCRIBE_MODE_NAME est défini sur subscribe, la valeur de INITIAL_CHECKPOINT_NAME ne prend effet que lors du premier démarrage du client d'abonnement.	L'heure de début de la consommation doit se situer dans la plage de données de l'instance d'abonnement et doit être convertie en horodatage UNIX. Remarque Vous pouvez afficher la plage de données de l'instance d'abonnement dans la colonne Data Range de la liste des tâches d'abonnement. Vous pouvez utiliser un moteur de recherche pour trouver un convertisseur d'horodatage UNIX.
   USE_CONFIG_CHECKPOINT_NAME	La valeur par défaut est true, ce qui force le client à démarrer à partir de l'heure spécifiée. Cela permet d'éviter la perte de données reçues mais non traitées.	Aucun
   SUBSCRIBE_MODE_NAME	Pour exécuter deux clients Kafka ou plus dans le même groupe de consommateurs, définissez ce paramètre sur subscribe sur tous les clients. La valeur par défaut est assign. Dans ce mode, cette fonctionnalité est désactivée et nous vous recommandons de déployer un seul client.	Aucun

9. Dans la barre de menus supérieure, sélectionnez Run > Run pour exécuter le client.

   Remarque

   Lors de la première exécution de l'application, IntelliJ IDEA peut mettre un certain temps à télécharger et installer automatiquement les packages de dépendances requis.

Résultats

Le client s'abonne avec succès aux données de modification de la base de données source.

Pour afficher les données de modification détaillées, supprimez les caractères // de la ligne //log.info(ret); à la ligne 25 du fichier NotifyDemoDB.java et exécutez à nouveau le client.

FAQ

• Q : Pourquoi dois-je suivre manuellement le point de contrôle de consommation de mon client ?

  R : Le point de contrôle de consommation enregistré par DTS indique le moment où DTS reçoit la validation du client Kafka, et non le moment où votre application termine le traitement des données. Si votre application ou votre client Kafka s'arrête de manière inattendue, l'utilisation de votre propre point de contrôle enregistré vous permet de reprendre la consommation exactement au point d'interruption, évitant ainsi à la fois la duplication et la perte de données.

Gérer le point de contrôle

1. Configurez le client d'abonnement pour qu'il écoute les basculements de cluster dans le module de collecte de données DTS.

   Pour ce faire, enregistrez un ClusterSwitchListener en tant qu'intercepteur de consommateur en définissant la propriété suivante pour le consommateur :

   properties.setProperty(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, ClusterSwitchListener.class.getName());

   Le code suivant présente un exemple d'implémentation de ClusterSwitchListener :

   public class ClusterSwitchListener implements ClusterResourceListener, ConsumerInterceptor {
       private final static Logger LOG = LoggerFactory.getLogger(ClusterSwitchListener.class);
       private ClusterResource originClusterResource = null;
       private ClusterResource currentClusterResource = null;
   
       public ConsumerRecords onConsume(ConsumerRecords records) {
           return records;
       }
   
       public void close() {
       }
   
       public void onCommit(Map offsets) {
       }
   
       public void onUpdate(ClusterResource clusterResource) {
           synchronized (this) {
               originClusterResource = currentClusterResource;
               currentClusterResource = clusterResource;
               if (null == originClusterResource) {
                   LOG.info("Cluster updated to " + currentClusterResource.clusterId());
               } else {
                   if (originClusterResource.clusterId().equals(currentClusterResource.clusterId())) {
                       LOG.info("Cluster not changed on update:" + clusterResource.clusterId());
                   } else {
                       LOG.error("Cluster changed");
                       throw new ClusterSwitchException("Cluster changed from " + originClusterResource.clusterId() + " to " + currentClusterResource.clusterId()
                               + ", consumer require restart");
                   }
               }
           }
       }
   
       public boolean isClusterResourceChanged() {
           if (null == originClusterResource) {
               return false;
           }
           if (originClusterResource.clusterId().equals(currentClusterResource.clusterId())) {
               return false;
           }
           return true;
       }
   
       public void configure(Map<String, ?> configs) {
       }
   
       public static class ClusterSwitchException extends KafkaException {
           public ClusterSwitchException(String message, Throwable cause) {
               super(message, cause);
           }
   
           public ClusterSwitchException(String message) {
               super(message);
           }
   
           public ClusterSwitchException(Throwable cause) {
               super(cause);
           }
   
           public ClusterSwitchException() {
               super();
           }
   
       }

2. Gérez l'événement de basculement de cluster du module de collecte de données DTS.

   Lorsqu'une exception ClusterSwitchException est interceptée, réinitialisez le point de contrôle initial pour le prochain abonnement à l'horodatage du dernier enregistrement consommé avec succès. L'extrait de code suivant fournit un exemple :

   try{
      //do some action
   } catch (ClusterSwitchListener.ClusterSwitchException e) {
      reset();
   }
   
   // Reset the checkpoint.
   public reset() {
     long offset = kafkaConsumer.offsetsForTimes(timestamp);
     kafkaConsumer.seek(tp,offset);
   }

   Remarque

   Pour un exemple d'implémentation, consultez KafkaRecordFetcher.

Correspondance entre le type de données et dataTypeNumber

Consommer les données d'abonnement à l'aide de Python

Si votre application de consommation est écrite en Python, vous pouvez utiliser les bibliothèques kafka-python et avro pour consommer et désérialiser les données au format Avro provenant du canal d'abonnement aux données. L'exemple suivant a été vérifié avec kafka-python 3.0.7 et avro 1.12.1.

1. Installez les bibliothèques Python requises.

   pip install kafka-python avro

2. Téléchargez le fichier de schéma Avro Record.avsc depuis le répertoire avro du dépôt de la démonstration du client Kafka. Il s'agit du même fichier de schéma référencé dans la section Notes de cette rubrique.

3. Utilisez l'exemple de code suivant pour créer un KafkaConsumer, charger le schéma Avro et désérialiser les données de modification :

   from kafka import KafkaConsumer
   import avro.schema
   from avro.io import DatumReader, BinaryDecoder
   
   schema = avro.schema.parse(open("Record.avsc", "rb").read())
   record_schema = schema.schemas[13]
   
   consumer = KafkaConsumer(
       "<topic>",
       bootstrap_servers="<DTS access point:port>",
       group_id="<consumer group ID>",
       enable_auto_commit=False,
       security_protocol="SASL_SSL",
       sasl_mechanism="PLAIN",
       sasl_plain_username="<consumer group account>",
       sasl_plain_password="<password>",
       ssl_ca_location="ca.pem",
   )
   
   for msg in consumer:
       reader = DatumReader(record_schema)
       record = reader.read(BinaryDecoder(msg.value))
       if record.get("operation") == "HEARTBEAT":
           continue
       print(record)

   L'exemple de code effectue les opérations suivantes :

   • Configure le KafkaConsumer pour se connecter en utilisant le transport SASL_SSL avec l'authentification PLAIN.

   • Charge Record.avsc et sélectionne la définition de schéma Record (schemas[13]) pour désérialiser chaque message avec DatumReader et BinaryDecoder.

   • Filtre les messages HEARTBEAT, qui sont des messages de maintien de connexion et ne contiennent pas de données de modification.

4. Remplacez les espaces réservés suivants par vos valeurs réelles. Vous pouvez obtenir ces valeurs en utilisant les mêmes méthodes décrites dans le tableau des paramètres de la section Procédure (KAFKA_TOPIC, KAFKA_BROKER_URL_NAME, SID_NAME, USER_NAME et PASSWORD_NAME) :

   Espace réservé	Description
   <topic>	Sujet d'abonnement du canal d'abonnement aux données. Correspond à KAFKA_TOPIC.
   <DTS access point:port>	Adresse réseau du canal d'abonnement aux données. Correspond à KAFKA_BROKER_URL_NAME.
   <consumer group ID>	ID du groupe de consommateurs. Correspond à SID_NAME.
   <consumer group account>	Nom d'utilisateur du groupe de consommateurs. Correspond à USER_NAME.
   <password>	Mot de passe du compte du groupe de consommateurs. Correspond à PASSWORD_NAME.