Data Transmission Service:Consommer les données de suivi à l'aide de flink-dts-connector

Remarques sur l'utilisation

• Le connecteur prend en charge les programmes Flink utilisant l'API DataStream, ainsi que l'API Table et SQL.

• Si votre programme Flink utilise l'API Table et SQL, vous ne pouvez le configurer que pour consommer les données d'une seule table à la fois. Pour consommer les données de plusieurs tables, vous devez configurer et exécuter une tâche distincte pour chaque table.

Procédure

L'exemple suivant utilise IntelliJ IDEA Community Edition 2020.1 pour Windows afin d'illustrer la consommation de données de suivi avec le connecteur flink-dts-connector.

1. Créez une instance de suivi des modifications. Pour plus d'informations, consultez Créer une instance de suivi des modifications pour une instance ApsaraDB RDS for MySQL, Créer une instance de suivi des modifications pour un cluster PolarDB for MySQL, ou Créer une instance de suivi des modifications pour une instance ApsaraDB for Oracle.

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

3. Téléchargez le fichier flink-dts-connector et décompressez-le.

4. Démarrez IntelliJ IDEA et cliquez sur Open or Import.

5. Dans la boîte de dialogue qui s'affiche, accédez au répertoire dans lequel vous avez décompressé le fichier flink-dts-connector. Développez les dossiers pour trouver le fichier Project Object Model (POM) : pom.xml.

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

7. 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>

8. Dans IntelliJ IDEA, développez les dossiers du projet et sélectionnez le fichier Java approprié en fonction du type d'API Flink utilisé par votre programme.

   • Si votre programme 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 effectuez les étapes suivantes :

     1. Dans la barre de menus supérieure, choisissez Run > Run....

     2. Dans la fenêtre contextuelle, cliquez sur Edit > DtsExample.

     3. Dans le champ Program arguments, saisissez les paramètres et leurs valeurs comme indiqué dans l'exemple suivant. Cliquez ensuite sur Run pour démarrer le connecteur flink-dts-connector.

        Remarque

        Pour plus de détails sur les paramètres et la manière de trouver leurs valeurs, consultez la section Paramètres.

        --broker-url dts-cn-******.******.***:****** --topic cn_hangzhou_rm_**********_dtstest_version2 --sid dts****** --user dtstest --password Test123456 --checkpoint 1624440043

     4. La sortie indique que le programme suit avec succès les modifications de données de la base de données source. Après le démarrage, les enregistrements de modification de données dans le DataStream ressemblent à l'exemple suivant.

        LazyParseRecord {operationType [HEARTBEAT], checkpoint [0@12006303@289540@2688@1625045211000]}
        LazyParseRecord {operationType [HEARTBEAT], checkpoint [0@12006307@290047@2688@1625045212000]}
        LazyParseRecord {operationType [UPDATE], checkpoint [0@12006305@290016@2688@1625045212000]}
        LazyParseRecord {operationType [HEARTBEAT], checkpoint [0@12006308@290047@2688@1625045214000]}
        LazyParseRecord {operationType [HEARTBEAT], checkpoint [0@12006309@290047@2688@1625045215000]}
        LazyParseRecord {operationType [HEARTBEAT], checkpoint [0@12006310@290047@2688@1625045216000]}
        LazyParseRecord {operationType [HEARTBEAT], checkpoint [0@12006311@290047@2688@1625045217000]}

        Remarque

        Pour afficher les détails des enregistrements de modification de données, connectez-vous à l'interface utilisateur Task Manager de votre programme Flink.

   • Si votre programme Flink utilise l'API Table et SQL, double-cliquez sur le fichier flink-dts-connector-master\src\test\java\com\alibaba\flink\connectors\dts\sql\DtsTableISelectTCaseTest.java et effectuez les étapes suivantes :

     Remarque

     Un seul fichier DtsTableISelectTCaseTest.java permet de configurer et de consommer les données de suivi d'une seule table. Pour consommer les données de plusieurs tables, vous devez configurer et exécuter une tâche distincte pour chaque table.

     1. Mettez en commentaire la ligne properties.load en ajoutant le préfixe //.

        /*Loads the parameter values that you set on the platform page into the Properties object.*/
        //properties.load(new StringReader(new String(Files.readAllBytes(Paths.get(configFilePath)), StandardCharsets.UTF_8)));

     2. Définissez la table unique à partir de laquelle vous souhaitez consommer les données.

     3. Définissez les paramètres de l'instance de suivi des modifications. Pour plus de détails sur les paramètres et la manière de trouver leurs valeurs, consultez la section Paramètres.

        public static void main(String[] args) throws Exception {
            setup(args);
            final String createTable =
                    "create table `dts` (\n"
                    + " `ts` TIMESTAMP(3) METADATA FROM 'timestamp',\n"
                    + " `id` bigint,\n"
                    + " `name` varchar,\n"
                    + " `age` bigint,\n"
                    + " WATERMARK FOR ts AS ts - INTERVAL '5' SECOND"
                    + ") with (\n"
                    + "'connector' = 'dts',"
                    + "'dts.server' = 'dts-cn-xxx:18001',"
                    + "'topic' = 'cn_hangzhou_rm_xxx_dtstest_version2',"
                    + "'dts.sid' = 'dtsxxx', "
                    + "'dts.user' = 'dtstest', "
                    + "'dts.password' = 'xxx',"
                    + "'dts.checkpoint' = '1624440043', "
                    + "'dts-cdc.table.name' = 'dtstestdata.order',"
                    + "'format' = 'dts-cdc')";
        }

     4. En haut de l'interface IntelliJ IDEA, cliquez sur Run'DtsTableISelectTCaseTest' pour démarrer le connecteur flink-dts-connector.

     5. La sortie indique que le programme suit avec succès les modifications de données de la base de données source. Après le démarrage, le terminal affiche les journaux de modifications (changelog). Les événements de mise à jour apparaissent par paires : la valeur avant la mise à jour (-U) et la valeur après la mise à jour (+U).

        ######> (false,-U(2021-06-23T20:32:17.391,null,null,null))
        ######> (true,+U(2021-06-23T20:32:17.391,null,null,null))
        ######> (false,-U(2021-06-23T20:32:45.604,null,null,null))
        ######> (true,+U(2021-06-23T20:32:45.604,null,null,null))
        ######> (false,-U(2021-06-30T17:26:52.201,null,null,null))
        ######> (true,+U(2021-06-30T17:26:52.201,null,null,null))
        ######> (false,-U(2021-06-30T19:19:26.975,null,null,null))
        ######> (true,+U(2021-06-30T19:19:26.975,null,null,null))

        Remarque

        Pour afficher les détails des enregistrements de modification de données, connectez-vous à l'interface utilisateur Task Manager de votre programme Flink.

Paramètres

FAQ

Cluster changed from *** to ***, consumer require restart.