Comment utiliser le connecteur flink-dts-connector - Data Transmission Service

Une fois l'instance d'abonnement configurée, utilisez le connecteur flink-dts-connector pour consommer les données de l'instance depuis un client Flink. Cette rubrique explique comment utiliser le connecteur flink-dts-connector.

Remarques sur l'utilisation

Ce connecteur prend en charge les clients Flink utilisant l'API DataStream, l'API Table ou SQL.
Si votre client Flink utilise l'API Table ou SQL, chaque configuration ne permet de consommer les données que d'une seule table. Pour consommer les données de plusieurs tables, configurez et exécutez plusieurs tâches indépendantes.

Procédure

Cette rubrique utilise IntelliJ IDEA Community Edition 2020.1 pour Windows afin de montrer comment utiliser le connecteur flink-dts-connector pour consommer les données d'une instance d'abonnement.

Créez une tâche de suivi des modifications. Pour plus d'informations, consultez les rubriques pertinentes dans Présentation du suivi des modifications.
Créez un ou plusieurs groupes de consommateurs. Pour plus d'informations, consultez Ajouter un groupe de consommateurs.
Téléchargez et décompressez le flink-dts-connector.
Ouvrez IntelliJ IDEA et cliquez sur Open or Import.
Dans la boîte de dialogue qui s'affiche, accédez au répertoire flink-dts-connector décompressé, développez les dossiers, puis recherchez le fichier Project Object Model (pom.xml).
Dans la boîte de dialogue qui s'affiche, sélectionnez Open as Project.

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

<dependency>
      <groupId>com.alibaba.flink</groupId>
      <artifactId>flink-dts-connector</artifactId>
      <version>1.1.1-SNAPSHOT</version>
      <classifier>jar-with-dependencies</classifier>
</dependency>

Dans IntelliJ IDEA, développez les dossiers et sélectionnez le fichier Java correspondant au type de programme de votre connecteur Flink.
- Si votre client Flink utilise l'API DataStream, double-cliquez sur le fichier flink-dts-connector-master\src\test\java\com\alibaba\flink\connectors\dts\datastream\DtsExample.java et suivez les étapes ci-dessous :
  1. En haut de la fenêtre IntelliJ IDEA, cliquez sur l'icône Run.
  2. Dans le menu qui s'affiche, cliquez sur DtsExample > Edit.
  3. Dans le champ Program arguments, saisissez les paramètres et leurs valeurs selon l'exemple suivant. Cliquez ensuite sur Run pour démarrer le connecteur flink-dts-connector.
    
    Remarque
    Pour plus d'informations sur les paramètres et sur la manière d'obtenir leurs valeurs, consultez Paramètres.
    --broker-url dts-cn-******.******.***:****** --topic cn_hangzhou_rm_**********_dtstest_version2 --sid dts****** --user dtstest --password Test123456 --checkpoint 1624440043
  4. La sortie confirme que le client peut s'abonner aux modifications de données de la base de données source.
    
    Remarque
    Pour interroger des enregistrements de modification de données spécifiques, connectez-vous à l'interface utilisateur Task Manager de votre client Flink.
- Si votre client Flink utilise l'API Table ou SQL, double-cliquez sur le fichier flink-dts-connector-master\src\test\java\com\alibaba\flink\connectors\dts\sql\DtsTableISelectTCaseTest.java et suivez les étapes ci-dessous :
  
  Remarque
  Chaque fichier DtsTableISelectTCaseTest.java ne peut être configuré que pour consommer les données d'une seule table. Pour configurer et consommer les données de plusieurs tables, répétez la configuration et exécutez plusieurs tâches indépendantes.
  1. Ajoutez deux barres obliques (//) pour mettre en commentaire la ligne de code spécifiée.
  2. Configurez les informations relatives à la table unique dont vous souhaitez consommer les données. Les instructions SQL sont prises en charge.
  3. Configurez les paramètres de l'instance d'abonnement. Pour plus d'informations sur les paramètres et sur la manière d'obtenir leurs valeurs, consultez Paramètres.
  4. En haut de la fenêtre IntelliJ IDEA, cliquez sur Run'DtsTableISelectTCaseTest' pour démarrer le connecteur flink-dts-connector.
  5. La sortie confirme que le client peut s'abonner aux modifications de données de la base de données source.
    
    Remarque
    Pour interroger des enregistrements de modification de données spécifiques, connectez-vous à l'interface utilisateur Task Manager de votre client Flink.

Paramètres

DtsExample.java	DtsTableISelectTCaseTest.java	Description	Méthode de requête
`broker-url`	`dts.server`	Le point de terminaison réseau et le port de l'instance d'abonnement. Remarque Si l'instance ECS sur laquelle votre client Flink est déployé et l'instance d'abonnement se trouvent dans le même réseau classique ou Virtual Private Cloud (VPC), nous vous recommandons d'utiliser un point de terminaison interne pour l'abonnement aux données afin de minimiser la latence réseau. Nous déconseillons l'utilisation d'un point de terminaison public en raison d'une éventuelle instabilité du réseau.	Dans la console DTS, cliquez sur l'ID de l'instance d'abonnement cible. Sur la page Basic Information, vous pouvez obtenir les informations relatives au Topic et au Network.
`topic`	`topic`	Le topic de l'instance d'abonnement.
`sid`	`dts.sid`	L'ID du groupe de consommateurs.	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. Vous pouvez obtenir l'Consumer Group ID/Name et le 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.
`user`	`dts.user`	Le compte du groupe de consommateurs. Avertissement Si vous n'utilisez pas le connecteur flink-dts-connector fourni dans cette rubrique, vous devez définir le nom d'utilisateur au format `<consumer_group_account>-<consumer_group_ID>`, par exemple `dtstest-dtsaebpv`. Sinon, la connexion échouera.
`password`	`dts.password`	Le mot de passe du compte.
`checkpoint`	`dts.checkpoint`	Le décalage du consommateur. Ce paramètre spécifie l'horodatage Unix du premier enregistrement de données à consommer. Exemple : 1624440043. Remarque Vous pouvez utiliser le décalage du consommateur pour : Reprendre la consommation à partir du dernier décalage après une interruption de l'application afin d'éviter toute perte de données. Spécifier un décalage pour commencer à consommer les données à partir d'un moment précis.	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 outil de conversion d'horodatage Unix.
N/A	`dts-cdc.table.name`	L'objet abonné. Vous ne pouvez spécifier qu'une seule table. Les exigences de format sont les suivantes : Si le Database Type est MySQL, PolarDB for MySQL, PolarDB-X 1.0 ou PolarDB-X 2.0, utilisez le format `<database_name>.<table_name>`. Pour les autres types de bases de données, utilisez le format `<schema_name>.<table_name>`.	Dans la console DTS, cliquez sur l'ID de l'instance d'abonnement cible. En haut de la page Basic Information ou Task Management, cliquez sur View Objects pour trouver la base de données et la table abonnées.

FAQ

Message d'erreur	Cause possible	Solution
`Cluster changed from * to *, consumer require restart.`	Un basculement dans le module DStore, utilisé par Data Transmission Service (DTS) pour lire les données incrémentielles, entraîne la perte du décalage du consommateur du client Flink.	Ne redémarrez pas le client. Interrogez plutôt le décalage du consommateur du client et transmettez à nouveau la valeur au paramètre `checkpoint` ou `dts.checkpoint` dans le fichier DtsExample.java ou DtsTableISelectTCaseTest.java pour reprendre la consommation des données.