Data Transmission Service:Consommer des données d’abonnement à l’aide d’un exemple de code SDK

Procédure

La procédure suivante montre comment exécuter l’exemple de code SDK pour consommer les données Data Subscription dans IntelliJ IDEA (Community Edition 2020.1 pour Windows).

1. Créez une instance Data Subscription. Pour plus d’informations, consultez Créer un canal Data Subscription pour une instance ApsaraDB RDS for MySQL, Créer un canal Data Subscription pour un cluster PolarDB for MySQL ou Créer un canal Data Subscription pour une base de données Oracle.

2. Créez un ou plusieurs groupes de consommateurs. Pour plus d’informations, consultez Créer des groupes de consommateurs.

   Important

   Lorsque vous consommez des données Data Subscription, vous devez appeler la méthode commit de DefaultUserRecord pour valider les points de contrôle. Le non-respect de cette étape peut entraîner une consommation en double des données.

3. Utilisez l’exemple de code SDK en fonction de vos besoins métier.

   • Utiliser le nouveau package SDK Data Subscription (recommandé)

     1. Ouvrez IntelliJ IDEA, puis cliquez sur Create New Project pour créer un projet pour votre application.

     2. Dans le projet, localisez le fichier POM (Project Object Model) : pom.xml.

     3. Ajoutez la dépendance suivante au fichier pom.xml :

        <dependency>
            <groupId>com.aliyun.dts</groupId>
            <artifactId>dts-new-subscribe-sdk</artifactId>
            <version>{dts_new_sdk_version}</version>
        </dependency>

        Remarque

        Vous pouvez trouver la dernière dépendance Maven sur la page dts-new-subscribe-sdk.

     4. Pour plus d’informations sur l’utilisation du nouveau SDK d’abonnement, consultez Utiliser l’exemple de code.

   • Utiliser une version personnalisée du nouveau SDK Data Subscription

     1. Téléchargez le package d’exemple de code SDK, puis décompressez-le.

        Remarque

        Cliquez sur et sélectionnez Download ZIP pour télécharger le package.

     2. Accédez au répertoire de l’exemple de code SDK décompressé. Utilisez un éditeur de texte pour ouvrir le fichier pom.xml et mettez à jour le SDK Data Subscription vers la dernière version.

        <name>dts-new-subscribe-sdk</name>
        <url>https://www.aliyun.com/product/dts</url>
        <description>The Aliyun new Subscribe SDK for Java used for accessing Data Transmission Service</description>
        <packaging>jar</packaging>
        <groupId>com.aliyun.dts</groupId>
        <artifactId>dts-new-subscribe-sdk</artifactId>
        <version>1.3</version>

        Important

        Vous pouvez obtenir la dernière version du SDK Data Subscription depuis le site web Maven. Pour plus d’informations, consultez la page Maven du SDK Data Subscription.

     3. Ouvrez IntelliJ IDEA. Sur l’écran d’accueil, cliquez sur Open or Import.

     4. Dans la boîte de dialogue Open File or Project, accédez au répertoire de l’exemple de code SDK décompressé, sélectionnez le fichier pom.xml, puis cliquez sur OK.

     5. Dans la boîte de dialogue qui s’affiche, sélectionnez Open as Project.

     6. Dans le panneau Project d’IntelliJ IDEA, développez src > test > java > com.aliyun.dts.subscribe.clients. Les éléments DTSConsumerAssignDemo et DTSConsumerSubscribeDemo sont les deux fichiers de démonstration d’exemple. Selon le mode d’utilisation du client SDK, sélectionnez et double-cliquez sur le fichier Java correspondant : DTSConsumerAssignDemo.java ou DTSConsumerSubscribeDemo.java.

        Arborescence des fichiers du projet :

        aliyun-dts-subscribe-sdk-java-master [dts-new-subscribe-sdk] D:\aliyun-dts-sub
          .idea
          src
            main
            test
              java
                com.aliyun.dts.subscribe.clients
                  DTSConsumerAssignDemo
                  DTSConsumerSubscribeDemo
                  UserMetaStore
          target
          .gitignore
          dts-new-subscribe-sdk.iml
          LICENSE
          pom.xml
          README.md
          External Libraries
          Scratches and Consoles

        Remarque

        DTS prend en charge les modes d’utilisation suivants du client SDK :

        • Mode ASSIGN : Afin de garantir l’ordre global des messages, DTS n’attribue qu’une seule partition (partition 0) à chaque sujet d’abonnement. Lors de l’utilisation d’un client SDK en mode ASSIGN, nous vous recommandons de ne démarrer qu’un seul client.

        • Mode SUBSCRIBE : Afin de garantir l’ordre global des messages, DTS n’attribue qu’une seule partition (partition 0) à chaque sujet d’abonnement. Si vous utilisez le client SDK en mode SUBSCRIBE, vous pouvez démarrer plusieurs clients SDK dans un groupe de consommateurs à des fins de reprise après sinistre. Si le client actif échoue, un autre client SDK est automatiquement attribué à la partition 0 pour reprendre la consommation.

4. Définissez les paramètres requis dans le fichier Java.

   public static void main(String[] args) {
       // kafka broker url
       String brokerUrl = "dts-cn-xxx18001";
       // topic to consume, partition is 0
       String topic = "cn_hangzhou_rm_xxx";
       // user password and sid for auth
       String sid = "dtsxxx";
       String userName = "dtstest";
       String password = "xxx";
       // initial checkpoint for first seek(a timestamp to set, eg 1566180200 if you want (Mon Aug 19 10:03:21 CST 2019))
       String initCheckpoint = "1620811813";
       // when use subscribe mode, group config is required. kafka consumer group is enabled
       ConsumerContext.ConsumerSubscribeMode subscribeMode = ConsumerContext.ConsumerSubscribeMode.ASSIGN;
       // if force use config checkpoint when start. for checkpoint reset, only assign mode works
       boolean isForceUseInitCheckpoint = true;

   Tableau 1. Paramètres obligatoires

   Paramètre	Description	Source
   brokerUrl	Endpoint et numéro de port de l'instance Data Subscription. Remarque Si l'instance ECS exécutant le client SDK et l'instance Data Subscription se trouvent dans le même réseau classique ou Virtual Private Cloud (VPC), nous vous recommandons d'utiliser l'endpoint interne pour l'abonnement afin de minimiser la latence réseau. Nous déconseillons l'utilisation d'un endpoint public en raison d'une éventuelle instabilité du réseau.	Dans la console DTS, cliquez sur l'ID de l'instance Data Subscription cible. Sur la page Basic Information, vous pouvez obtenir l'endpoint et le numéro de port dans la section Network.
   topic	Rubrique d'abonnement de l'instance.	Dans la console DTS, cliquez sur l'ID de l'instance Data Subscription cible. Sur la page Basic Information, vous pouvez obtenir la rubrique Topic dans la section Basic Information.
   sid	ID du groupe de consommateurs.	Dans la console DTS, cliquez sur l'ID de l'instance Data Subscription cible, puis choisissez Consume Data. Vous pouvez obtenir l'Consumer Group ID et le Account du groupe de consommateurs. Remarque Le mot de passe associé au nom d'utilisateur du groupe de consommateurs est défini lors de la création du groupe.
   userName	Nom d'utilisateur du groupe de consommateurs. Avertissement Si vous n'utilisez pas le client fourni dans cette rubrique, vous devez définir le nom d'utilisateur au format <Username>-<Consumer Group ID>. Exemple : dtstest-dtsaebpv. Dans le cas contraire, la connexion échouera.
   password	Mot de passe associé au nom d'utilisateur.
   initCheckpoint	Point de contrôle de consommation, spécifié sous forme d'horodatage UNIX, à partir duquel le client SDK commence à consommer les données. Exemple : 1620962769. Remarque Vous pouvez utiliser les informations de point de contrôle de consommation dans les scénarios suivants : Pour reprendre la consommation et éviter toute perte de données après une interruption de l'application, transmettez le dernier point de contrôle de consommation connu. Lors du démarrage du client, vous pouvez transmettre un point de contrôle de consommation spécifique pour consommer les données à partir d'une position souhaitée.	Le point de contrôle de consommation doit se situer dans la plage de données de l'instance Data Subscription (comme illustré dans la figure) et doit être converti en horodatage UNIX. Remarque Vous pouvez utiliser un moteur de recherche pour trouver un convertisseur d'horodatage UNIX.
   ConsumerContext.ConsumerSubscribeMode subscribeMode	Mode d'utilisation du client SDK. Valeurs possibles : ConsumerContext.ConsumerSubscribeMode.ASSIGN : mode ASSIGN. Un seul client SDK d'un groupe de consommateurs peut consommer les données Data Subscription. ConsumerContext.ConsumerSubscribeMode.SUBSCRIBE : mode SUBSCRIBE. Vous pouvez démarrer plusieurs clients SDK au sein du même groupe de consommateurs à des fins de reprise après sinistre.	S/O

5. Dans la barre de navigation supérieure d'IntelliJ IDEA, choisissez Run > Run pour exécuter le client.

   Remarque

   Lors de la première exécution du client, le chargement et l'installation automatiques des dépendances requises peuvent prendre un certain temps.

   • Après l'exécution, le client SDK consomme avec succès les données de modification de la base de données source. Voici un exemple d'enregistrement UPDATE consommé :

     [2021-05-18 16:49:50,260] INFO
     RecordID [559686]
     RecordTimestamp [1621327772]
     Source [{"sourceType": "MySQL", "version": "8.0.18"}]
     RecordType [UPDATE]
     Schema info [{,
     recordFields= [{fieldName='orderid', rawDataTypeNum=3, isPrimaryKey=true, isUniqueKey=false, fieldPosition=0}, {fieldName='username', rawDataTypeNum=254, isPrimaryKey=false, isUniqueKey=false, fieldPosition=1}, {fieldName='ordertime', rawDataTypeNum=12, isPrimaryKey=false, isUniqueKey=false, fieldPosition=2}, {fieldName='commodity', rawDataTypeNum=253, isPrimaryKey=false, isUniqueKey=false, fieldPosition=3}, {fieldName='phonenumber', rawDataTypeNum=3, isPrimaryKey=false, isUniqueKey=false, fieldPosition=4}, {fieldName='address', rawDataTypeNum=15, isPrimaryKey=false, isUniqueKey=false, fieldPosition=5}],
     databaseName='dtstestdata',
     tableName='order',
     primaryIndexInfo [{indexType=PrimaryKey, indexFields=[{fieldName='orderid', rawDataTypeNum=3, isPrimaryKey=true, isUniqueKey=false, fieldPosition=0}], cardinality=0, nullable=true, isFirstUniqueIndex=false, name=null}],
     uniqueIndexInfo [[]],
     partitionFields = null}]
     Before image {[Field [orderid] [1]
     Field [username] [ Jane  McLenahan]
     Field [ordertime] [xxx]
     Field [commodity] [xxx]
     Field [phonenumber] [ xxx]
     Field [address] [xxx]
     ]}
     After image {[Field [orderid] [1]
     Field [username] [ Jane]
     Field [ordertime] [xxx]
     Field [commodity] [xxx]
     Field [phonenumber] [ xxx]
     Field [address] [xxx]
     ]}
       (com.aliyun.dts.subscribe.clients.recordprocessor.DefaultRecordPrintListener)

   • Le client SDK agrège et affiche périodiquement des statistiques relatives à la consommation de données, notamment le nombre total et le volume des enregistrements envoyés et reçus, ainsi que le nombre de requêtes par seconde (RPS).

     [2021-05-18 16:25:09,167] INFO {"outCounts":488616.0,"outBytes":48606134,"outRps":1.15,"outBps":114.57,"count":11.0,"inBytes":60118961,"DStoreRecordQueue":0.0,"inCounts":557154.0,"inRps":1.12,"inBps":112.44,"__dt":1621326309167,"DefaultUserRecordQueue":0.0} (log_metrics)

     Tableau 2. Statistiques de consommation des données

     description	Description
     outCounts	Nombre total d'enregistrements de données consommés par le client SDK.
     outBytes	Volume total de données consommé par le client SDK. Unité : octets.
     outRps	Nombre de requêtes par seconde (RPS) auquel le client SDK consomme les données.
     outBps	Débit de consommation des données du client SDK, en bits par seconde (bps).
     inBytes	Volume total de données envoyé par le serveur DTS. Unité : octets.
     DStoreRecordQueue	Taille de la file d'attente du cache interne pour les enregistrements entrants provenant du serveur DTS.
     inCounts	Nombre total d'enregistrements de données envoyés par le serveur DTS.
     inRps	RPS auquel le serveur DTS envoie les données.
     __dt	Horodatage de réception des données par le client SDK. Unité : millisecondes.
     DefaultUserRecordQueue	Taille de la file d'attente contenant les enregistrements prêts à être traités par l'application consommatrice.

Enregistrement et interrogation des points de contrôle de consommation

Pour démarrer ou reprendre la consommation des données (par exemple, lors du premier lancement, d'un redémarrage ou d'une nouvelle tentative interne), le client SDK a besoin d'un point de contrôle de consommation. Le tableau suivant décrit comment gérer et interroger les points de contrôle dans différents scénarios afin d'éviter la perte de données, de minimiser les consommations en double et d'activer la consommation à la demande.

Persistance des points de contrôle de consommation

Lors d'un événement de reprise après sinistre dans le module de collecte incrémentale des données (en particulier en mode SUBSCRIBE), le nouveau module ne conserve pas le dernier point de contrôle de consommation du client. Le client peut alors reprendre à partir d'un point de contrôle plus ancien, ce qui entraîne une consommation en double des données historiques. Par exemple, avant le basculement, la plage de points de contrôle de l'ancien module s'étend du 11 novembre 2023 à 08:00:00 au 12 novembre 2023 à 08:00:00, et le point de contrôle du client est fixé au 12 novembre 2023 à 08:00:00. Après le basculement, la plage de points de contrôle du nouveau module va du 8 novembre 2023 à 10:00:00 au 12 novembre 2023 à 08:01:00. Le client démarre à partir du point de contrôle initial du nouveau module (8 novembre 2023 à 10:00:00), provoquant ainsi une consommation en double.

Pour éviter la consommation en double dans ce scénario, configurez un magasin de points de contrôle persistant sur le client. L'exemple suivant présente une implémentation possible que vous pouvez adapter selon vos besoins.

1. Créez une classe UserMetaStore qui étend AbstractUserMetaStore.

   Par exemple, pour stocker les informations de point de contrôle dans une base de données MySQL, utilisez le code Java suivant :

   public class UserMetaStore extends AbstractUserMetaStore {
   
       @Override
       protected void saveData(String groupID, String toStoreJson) {
           Connection con = getConnection();
   			  String sql = "insert into dts_checkpoint(group_id, checkpoint) values(?, ?)";
   
           PreparedStatement pres = null;
           ResultSet rs = null;
   
           try {
               pres = con.prepareStatement(sql);
               pres.setString(1, groupID);
               pres.setString(2, toStoreJson);
               pres.execute();
           } catch (Exception e) {
               e.printStackTrace();
           } finally {
               close(rs, pres, con);
           }
       }
   
       @Override
       protected String getData(String groupID) {
           Connection con = getConnection();
   			  String sql = "select checkpoint from dts_checkpoint where group_id = ?";
   
           PreparedStatement pres = null;
           ResultSet rs = null;
   
           try {
               pres = con.prepareStatement(sql);
               pres.setString(1, groupID);
   						ResultSet rs = pres.executeQuery()
                             
               String checkpoint = rs.getString("checkpoint");
             
               return checkpoint;
           } catch (Exception e) {
               e.printStackTrace();
           } finally {
               close(rs, pres, con);
           }
       }
   }
   

2. Dans le fichier consumerContext.java, configurez le support de stockage externe à l'aide de la méthode setUserRegisteredStore(new UserMetaStore()).

FAQ

• Comment résoudre les problèmes de connexion avec une instance Data Subscription ?

  Dépannez le problème en vous basant sur le message d'erreur. Pour plus d'informations, consultez la section Dépannage.

• Sous quel format les points de contrôle de consommation sont-ils persistés ?

  Les données de point de contrôle de consommation persistées sont stockées au format JSON. Le point de contrôle persisté est un horodatage UNIX qui peut être transmis directement au SDK. Dans l'exemple de réponse ci-dessous, la valeur 1700709977 associée à la clé "timestamp" correspond au point de contrôle de consommation persisté.

  {"groupID":"dtsglg11d48230***","streamCheckpoint":[{"partition":0,"offset":577989,"topic":"ap_southeast_1_vpc_rm_t4n22s21iysr6****_root_version2","timestamp":1700709977,"info":""}]}

Dépannage

ERROR
CheckResult{isOk=false, errMsg='telnet dts-cn-hangzhou.aliyuncs.com:18009
failed, please check the network and if the brokerUrl is correct'}
(com.aliyun.dts.subscribe.clients.DefaultDTSConsumer)

telnet real node *** failed, please check the network

ERROR CheckResult{isOk=false, errMsg='build kafka consumer failed, error: org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata, probably the user name or password is wrong'} (com.aliyun.dts.subscribe.clients.DefaultDTSConsumer)

com.aliyun.dts.subscribe.clients.exception.TimestampSeekException: RecordGenerator:seek timestamp for topic [cn_hangzhou_rm_bp11tv2923n87081s_rdsdt_dtsacct-0] with timestamp [1610249501] failed