MaxCompute:Flink CDC open source pour l'écriture quasi temps réel dans les tables Delta

Configurer les environnements

Soumettre un travail de pipeline YAML

1. Téléchargez les packages JAR requis :

   • Package Flink CDC : téléchargez le fichier flink-cdc-3.1.1-bin.tar.gz et décompressez-le pour obtenir le répertoire flink-cdc-3.1.1. Déplacez le contenu des répertoires bin, lib, log et conf vers les répertoires correspondants sous flink-1.18.0.

   • Packages de connecteurs : téléchargez les fichiers JAR suivants et placez-les dans le répertoire flink-1.18.0/lib : > Remarque : les liens de téléchargement ci-dessus pointent uniquement vers des versions publiées. Pour utiliser une version SNAPSHOT, compilez le code source à partir de la branche master ou release localement.

     • Connecteur de pipeline MySQL 3.1.1

     • Connecteur de pipeline MaxCompute 3.1.1

   • Pilote JDBC : téléchargez le fichier mysql-connector-java-8.0.27.jar. Transmettez-le à l'interface CLI Flink CDC avec l'option --jar, ou placez-le dans $FLINK_HOME/lib et redémarrez le cluster. Les connecteurs CDC n'incluent plus les pilotes JDBC.

2. Créez le fichier YAML du pipeline. L'exemple suivant (mysql-to-maxcompute.yaml) synchronise toutes les tables de la base de données app_db :

   Espace réservé	description
   ${your_accessId}	Votre ID AccessKey. Obtenez-le depuis la page Paire de clés d'accès.
   ${your_accessKey}	Le secret AccessKey correspondant à votre ID AccessKey.
   ${your_maxcompute_endpoint}	L'endpoint MaxCompute pour votre région et votre réseau. Consultez la rubrique Endpoints.
   ${your_project}	Le nom de votre projet MaxCompute. Retrouvez-le dans la console MaxCompute sous Workspace > Projects.

   ################################################################################
   # Description: Sync MySQL all tables to MaxCompute
   ################################################################################
   source:
     type: mysql
     hostname: localhost
     port: 3306
     username: root
     password: 123456
     tables: app_db\.*
     server-id: 5400-5404
     server-time-zone: UTC
   
   # Configure the accessId, accessKey, endpoint, and project parameters.
   sink:
     type: maxcompute
     name: MaxComputeSink
     accessId: ${your_accessId}
     accessKey: ${your_accessKey}
     endpoint: ${your_maxcompute_endpoint}
     project: ${your_project}
     bucketsNum: 8
   
   pipeline:
     name: Sync MySQL Database to MaxCompute
     parallelism: 1

   Remplacez les espaces réservés suivants : Pour connaître tous les paramètres source disponibles, consultez la rubrique Connecteur MySQL. Pour connaître tous les paramètres de destination disponibles, consultez la section Configuration du connecteur.

3. Soumettez le travail de pipeline au cluster Flink autonome :

   ./bin/flink-cdc.sh mysql-to-maxcompute.yaml

   Une soumission réussie renvoie une sortie similaire à :

   Pipeline has been submitted to cluster.
   Job ID: f9f9689866946e25bf151ecc179ef46f
   Job Description: Sync MySQL Database to MaxCompute

   L'interface web Flink affiche un travail en cours d'exécution nommé Sync MySQL Database to MaxCompute.

4. Dans MaxCompute, vérifiez que les tables sont créées et que les données sont écrites :

   -- Query the orders table.
   read orders;
   
   -- Expected result:
   +------------+------------+
   | id         | price      |
   +------------+------------+
   | 1          | 4          |
   | 2          | 100        |
   +------------+------------+
   
   -- Query the shipments table.
   read shipments;
   
   -- Expected result:
   +------------+------------+
   | id         | city       |
   +------------+------------+
   | 1          | beijing    |
   | 2          | xian       |
   +------------+------------+
   
   -- Query the products table.
   read products;
   
   -- Expected result:
   +------------+------------+
   | id         | product    |
   +------------+------------+
   | 3          | Peanut     |
   | 1          | Beer       |
   | 2          | Cap        |
   +------------+------------+

Synchroniser les modifications en temps réel

Lorsque le pipeline est en cours d'exécution, les modifications effectuées dans MySQL sont reflétées dans MaxCompute en temps réel. Les exemples suivants utilisent la table orders pour illustrer les opérations d'insertion, de modification du schéma, de mise à jour et de suppression.

1. Connectez-vous au conteneur MySQL :

   docker-compose exec mysql mysql -uroot -p123456

2. Insérez une ligne dans la table orders :

   INSERT INTO app_db.orders (id, price) VALUES (3, 100.00);

   Exécutez read orders; dans MaxCompute. La nouvelle ligne apparaît :

   +------------+------------+
   | id         | price      |
   +------------+------------+
   | 3          | 100        |
   | 1          | 4          |
   | 2          | 100        |
   +------------+------------+

3. Ajoutez une colonne à la table orders dans MySQL :

   ALTER TABLE app_db.orders ADD amount varchar(100) NULL;

   Exécutez read orders; dans MaxCompute. La nouvelle colonne amount est propagée automatiquement :

   +------------+------------+------------+
   | id         | price      | amount     |
   +------------+------------+------------+
   | 3          | 100        | NULL       |
   | 1          | 4          | NULL       |
   | 2          | 100        | NULL       |
   +------------+------------+------------+

4. Mettez à jour une ligne dans la table orders :

   UPDATE app_db.orders SET price=100.00, amount=100.00 WHERE id=1;

   Exécutez read orders; dans MaxCompute :

   +------------+------------+------------+
   | id         | price      | amount     |
   +------------+------------+------------+
   | 3          | 100        | NULL       |
   | 1          | 100        | 100.00     |
   | 2          | 100        | NULL       |
   +------------+------------+------------+

5. Supprimez une ligne de la table orders :

   DELETE FROM app_db.orders WHERE id=2;

   Exécutez read orders; dans MaxCompute :

   +------------+------------+------------+
   | id         | price      | amount     |
   +------------+------------+------------+
   | 3          | 100        | NULL       |
   | 1          | 100        | 100.00     |
   +------------+------------+------------+

Router les tables

Flink CDC permet de router les schémas et les données des tables sources vers différents noms de tables de destination. Utilisez cette fonctionnalité pour les migrations de base de données, les renommages de tables ou les fusions de bases de données fragmentées.

L'exemple YAML suivant route chaque table source vers une table cible portant un nom différent :

################################################################################
# Description: Sync MySQL all tables to MaxCompute
################################################################################
source:
  type: mysql
  hostname: localhost
  port: 3306
  username: root
  password: 123456
  tables: app_db\.*
  server-id: 5400-5404
  server-time-zone: UTC

# Configure the accessId, accessKey, endpoint, and project parameters.
sink:
  type: maxcompute
  name: MaxComputeSink
  accessId: ${your_accessId}
  accessKey: ${your_accessKey}
  endpoint: ${your_maxcompute_endpoint}
  project: ${your_project}
  bucketsNum: 8

route:
  - source-table: app_db.orders
    sink-table: ods_db.ods_orders
  - source-table: app_db.shipments
    sink-table: ods_db.ods_shipments
  - source-table: app_db.products
    sink-table: ods_db.ods_products

pipeline:
  name: Sync MySQL Database to MaxCompute
  parallelism: 1

Le champ source-table prend en charge les expressions régulières. Pour fusionner plusieurs tables fragmentées en une seule table cible :

route:
  - source-table: app_db.order\.*
    sink-table: ods_db.ods_orders

Cette configuration agrège les données de app_db.order01, app_db.order02, app_db.order03 et des tables similaires dans ods_db.ods_orders.

Les scénarios dans lesquels des valeurs de clé primaire en double existent entre plusieurs tables sources ne sont pas pris en charge et le seront dans les prochaines versions du connecteur.

Pour connaître tous les paramètres de routage, consultez la rubrique Route.

Nettoyer l'environnement

Une fois le tutoriel terminé, arrêtez tous les composants en cours d'exécution.

1. Arrêtez les conteneurs Docker. Exécutez la commande suivante dans le répertoire contenant le fichier docker-compose.yaml :

   docker-compose down

2. Arrêtez le cluster Flink. Exécutez la commande suivante dans le répertoire flink-1.18.0 :

   ./bin/stop-cluster.sh

Configuration du connecteur

Le connecteur de destination MaxCompute prend en charge les paramètres de configuration suivants.

Les données sont vidées vers MaxCompute lorsque le tampon d'une partition ou d'un compartiment dépasse le seuil de taille configuré, ou lorsqu'un point de contrôle (checkpoint) est déclenché.

Mappages d'emplacement de table

Lorsque le connecteur crée automatiquement une table, il mappe les informations d'emplacement de la source vers MaxCompute comme suit.

Mappages de types de données

Le tableau suivant montre comment les types de données Flink sont mappés vers les types de données MaxCompute.