Créer une tâche d'ingestion de base de données en temps réel - Realtime Compute for Apache Flink

Realtime Compute for Apache Flink simplifie l'ingestion de données en temps réel en gérant automatiquement le basculement de la synchronisation complète vers la synchronisation incrémentielle, la découverte des métadonnées, l'évolution du schéma et la synchronisation de base de données entière. Cette rubrique explique comment créer rapidement une tâche d'ingestion de données qui diffuse les données d'ApsaraDB RDS for MySQL vers Hologres.

Contexte

L'illustration suivante montre l'apparence de ces bases de données et tables dans la console DMS.

L'instance RDS contient plusieurs bases de données (tpc_ds, tpc_ds_large, user_db1, user_db2, user_db3, __recycle_bin__). La base de données user_db3 contient trois tables : user03, user06 et user09. La table user03 comporte deux colonnes : id (int(11)) et name (varchar(255)).

Suivez les étapes ci-dessous pour développer une tâche d'ingestion de données qui synchronise toutes ces tables vers Hologres et fusionne les tables utilisateur fragmentées en une seule table :

Étape 1 : Développer une tâche d'ingestion de données
Étape 2 : Démarrer la tâche
Étape 3 : Vérifier les résultats de la synchronisation complète
Étape 4 : Vérifier les résultats de la synchronisation incrémentielle
(Facultatif) Étape 5 : Configurer les ressources de la tâche

Cette rubrique utilise l'ingestion de données Flink CDC pour effectuer la synchronisation de base de données entière et fusionner les tables fragmentées. Cela vous permet de réaliser la synchronisation complète et incrémentielle des données, ainsi que la synchronisation des modifications de schéma en temps réel, avec une seule tâche.

Prérequis

Si vous utilisez un utilisateur RAM ou un rôle RAM, assurez-vous de disposer des autorisations requises pour accéder à la console Flink. Pour plus d'informations, consultez Gérer les autorisations.
Vous avez créé un espace de travail Flink. Pour plus d'informations, consultez Activer Realtime Compute for Apache Flink.
Stockage amont et aval
- Vous avez créé une instance ApsaraDB RDS for MySQL. Pour plus d'informations, consultez (Obsolète, redirection vers « Étape 1 ») Créer rapidement une instance ApsaraDB RDS for MySQL.
- Vous avez créé une instance Hologres. Pour plus d'informations, consultez Acheter une instance Hologres.
Remarque
Les instances ApsaraDB RDS for MySQL et Hologres doivent se trouver dans la même région et le même réseau privé virtuel (VPC) que l'espace de travail Flink. Dans le cas contraire, vous devez établir une connexion réseau. Pour plus d'informations, consultez Comment accéder à d'autres services via des VPC ? et Comment accéder à Internet ?.
Vous avez préparé les données de test et configuré les listes d'autorisation IP. Pour plus d'informations, consultez Préparer les données de test MySQL et une base de données Hologres et Configurer une liste d'autorisation IP.

Préparer les données de test MySQL et une base de données Hologres

Cliquez sur tpc_ds.sql, user_db1.sql, user_db2.sql et user_db3.sql pour télécharger les fichiers de données de test sur votre machine locale.
Dans la console DMS, préparez les données de test dans votre instance ApsaraDB RDS for MySQL.
1. Connectez-vous à votre instance ApsaraDB RDS for MySQL via DMS.
  
  Pour plus d'informations, consultez (Obsolète, redirection vers « Étape 2 ») Se connecter à une instance ApsaraDB RDS for MySQL via DMS.
2. Dans la fenêtre SQL Console, saisissez les commandes suivantes et cliquez sur Execute.
  Les commandes suivantes créent quatre bases de données : tpc_ds, user_db1, user_db2 et user_db3.
```
CREATE DATABASE tpc_ds;
CREATE DATABASE user_db1;
CREATE DATABASE user_db2;
CREATE DATABASE user_db3;
```
3. Dans la barre de navigation supérieure, cliquez sur Data Import.
4. Sous l'onglet Batch Data Import, sélectionnez la base de données dans laquelle importer les données, téléchargez le fichier SQL correspondant, cliquez sur Submit, puis sur Execute Change. Dans la boîte de dialogue qui s'affiche, cliquez sur Confirm Execution.
  
  Répétez cette opération pour importer les fichiers de données correspondants dans les bases de données tpc_ds, user_db1, user_db2 et user_db3.
  
  Définissez File Encoding sur Auto Detect, Import Mode sur Express Mode ou Safe Mode, et File Type sur SQL Script, CSV ou Excel. Les pièces jointes prennent en charge les formats txt/sql/csv/xlsx/zip, jusqu'à 5 Go.
Dans la console Hologres, créez une base de données nommée my_user pour stocker les données de la table utilisateur fusionnée.

Pour plus d'informations, consultez Créer une base de données.

Configurer une liste d'autorisation IP

Pour permettre à Flink d'accéder aux instances ApsaraDB RDS for MySQL et Hologres, ajoutez le bloc CIDR de votre espace de travail Flink à la liste d'autorisation IP des deux instances.

Obtenez le bloc CIDR de l'espace de travail Flink.
1. Connectez-vous à la console Realtime Compute for Apache Flink.
2. Dans la liste des espaces de travail, repérez l'workspace cible et choisissez More > Workspace Details dans la colonne Actions.
3. Dans la boîte de dialogue Workspace Details, consultez les informations relatives au CIDR Block du vSwitch Flink.
Ajoutez le bloc CIDR Flink à la liste d'autorisation IP de l'instance ApsaraDB RDS for MySQL.

Pour plus d'informations, consultez Configurer une liste d'autorisation IP.

Dans la boîte de dialogue Modify Whitelist Group, saisissez le bloc CIDR entièrement géré par Flink dans le champ Group Whitelist, puis cliquez sur OK.
Ajoutez le bloc CIDR Flink à la liste d'autorisation IP de l'instance Hologres.
Lors de la configuration d'une connexion de données dans HoloWeb, définissez Login Method sur Password-free login for current user avant de configurer une liste d'autorisation IP pour la connexion. Pour plus d'informations, consultez Liste d'autorisation IP.

Dans HoloWeb, choisissez Security Center dans la barre de navigation supérieure, cliquez sur IP Whitelist dans le volet de gauche, et configurez les paramètres suivants dans la boîte de dialogue Edit IP Whitelist :
- Group : Sélectionnez default
- Database Restriction : Sélectionnez ALL
- User Restriction : Sélectionnez ALL
- IP Address : Saisissez le bloc CIDR Flink en notation CIDR (par exemple, 172.xx.0/19). Séparez plusieurs adresses IP par des sauts de ligne.
Cliquez sur OK pour appliquer.

Étape 1 : Développer la tâche d'ingestion

Connectez-vous à la console de développement Flink et créez une nouvelle tâche.

Sur la page Development > Data Ingestion, cliquez sur New.
Cliquez sur Blank Data Ingestion Draft.

Realtime Compute for Apache Flink propose un ensemble riche de modèles de code, chacun avec des cas d'utilisation spécifiques, des exemples de code et des conseils. Vous pouvez cliquer sur un modèle pour découvrir les fonctionnalités du produit et la syntaxe permettant d'implémenter votre logique métier.
Cliquez sur Next.

Dans la boîte de dialogue New Draft, configurez les paramètres.

Parameter	Description	Example
Name	Le nom de la tâche. Remarque Le nom de la tâche doit être unique au sein du projet actuel.	flink-test
Location	Le dossier dans lequel est stocké le fichier de code de la tâche. Vous pouvez également cliquer sur l'icône à côté d'un dossier existant pour créer un sous-dossier.	Job Drafts
Engine Version	La version du moteur Flink utilisée par la tâche. Pour obtenir des informations sur les numéros de version du moteur, la compatibilité des versions et les dates importantes du cycle de vie, consultez Versions du moteur.	vvr-11,1-jdk11-flink-1,20

Cliquez sur OK.

Copiez le code de tâche suivant dans l'éditeur de tâche.

Le code suivant synchronise toutes les tables de la base de données tpc_ds vers Hologres et fusionne les tables utilisateur fragmentées en une seule table dans Hologres :

source:
  type: mysql
  name: MySQL Source
  hostname: localhost
  port: 3306
  username: username
  password: password
  tables: tpc_ds.\.*,user_db[0-9]+.user[0-9]+
  server-id: 8601-8604
  # (Optional) Synchronize table and column comments.
  include-comments.enabled: true
  # (Optional) Prioritize the distribution of unbounded chunks to prevent potential TaskManager OutOfMemory errors.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Enable parsing filters to accelerate reading.
  scan.only.deserialize.captured.tables.changelog.enabled: true  

sink:
  type: hologres
  name: Hologres Sink
  endpoint: ****.hologres.aliyuncs.com:80
  dbname: cdcyaml_test
  username: ${secret_values.holo-username}
  password: ${secret_values.holo-password}
  sink.type-normalize-strategy: BROADEN
  
route:
  # Merge and synchronize the sharded user tables to the my_user.users table.
  - source-table: user_db[0-9]+.user[0-9]+
    sink-table: my_user.users

Remarque

Les tables de la base de données MySQL tpc_ds sont mappées directement vers des tables portant le même nom dans la destination, de sorte que la section route ne nécessite aucune configuration de mappage supplémentaire. Pour synchroniser les tables vers une base de données portant un nom différent, telle que ods_tps_ds, configurez le module route comme suit :

route:
  # Merge and synchronize the sharded user tables to the my_user.users table.
  - source-table: user_db[0-9]+.user[0-9]+
    sink-table: my_user.users
  # Rename the database for all tables under tpc_ds and synchronize them to ods_tps_ds.
  - source-table: tpc_ds.\.*
    sink-table: ods_tps_ds.<>
    replace-symbol: <>

Étape 2 : Démarrer la tâche

Sur la page Development > Data Ingestion, cliquez sur Deploy. Dans la boîte de dialogue qui s'affiche, cliquez sur Confirm.

Dans la boîte de dialogue, vous pouvez saisir des Remarks, définir des Job Tags et sélectionner une file d'attente dans la liste déroulante Deployment Target (par défaut : default-queue). Remarque : Le déploiement prend effet au prochain démarrage de la tâche.
Sur la page O&M > Deployments, cliquez sur Start dans la colonne Actions pour la tâche cible. Configurez les paramètres selon vos besoins. Pour plus d'informations, consultez Démarrer une tâche.
Cliquez sur Start.

Une fois la tâche démarrée, vous pouvez consulter ses informations d'exécution et son statut sur la page Job Operations. Les statuts de tâche incluent Running, Failed et Stopped. Utilisez le filtre déroulant Stream Jobs en haut pour filtrer la liste des tâches par type.

Étape 3 : Vérifier la synchronisation complète

Connectez-vous à la console de gestion Hologres.
Sous l'onglet Metadata Management, vérifiez que les 24 tables et leurs données sont présentes dans la base de données tpc_ds de l'instance Hologres.

Accédez à Instance > tpc_ds > public > Tables pour voir des tables telles que call_center, catalog_sales, customer et store_sales. Sélectionnez la table store_sales et cliquez sur l'onglet Data Preview pour afficher des colonnes telles que ss_sold_date, ss_sold_time, ss_item_sk et ss_customer.
Sous l'onglet Metadata Management, vérifiez le schéma de la table users dans la base de données my_user.
Les éléments suivants décrivent le schéma de table et les données synchronisés.
- Schéma de table
  
  Après la synchronisation, la table users contient les colonnes suivantes :
  - _db_name (text, clé primaire)
  - _table_name (text, clé primaire)
  - id (int4, clé primaire)
  - name (varchar 255, nullable)
  Les colonnes _db_name, _table_name et id forment une clé primaire composite.
  
  Le schéma de la table users inclut deux colonnes supplémentaires absentes des tables source MySQL : _db_name et _table_name. Ces colonnes indiquent la base de données et la table source pour chaque ligne et font partie de la clé primaire composite, garantissant l'unicité des données après la fusion des tables fragmentées.
- Données de table
  Dans le coin supérieur droit de la page d'informations de la table users, cliquez sur Query Table. Saisissez la commande suivante, puis cliquez sur Run.
  select * from users order by _db_name,_table_name,id;
  Les résultats de la requête montrent que les données utilisateur ont été entièrement synchronisées vers trois bases de données : user_db1 (user01, user04, user07), user_db2 (user02, user05, user08) et user_db3 (user03, user06, user09), soit un total de 9 enregistrements avec les colonnes _db_name, _table_name, id et name.

Étape 4 : Vérifier la synchronisation incrémentielle

Une fois la synchronisation complète terminée, la tâche passe automatiquement à la phase de synchronisation incrémentielle sans intervention manuelle. Vous pouvez vérifier la valeur currentEmitEventTimeLag sous l'onglet Monitoring and Alerts pour déterminer la phase de synchronisation des données.

Connectez-vous à la console Realtime Compute for Apache Flink.
Cliquez sur Console dans la colonne Actions pour l'espace de travail cible.
Sur la page O&M > Deployments, cliquez sur le nom de la tâche cible.
Cliquez sur l'onglet Alarm (ou Metrics).
Examinez le graphique currentEmitEventTimeLag pour déterminer la phase de synchronisation des données.
- Une valeur de 0 indique la phase de synchronisation complète.
- Une valeur supérieure à 0 indique la phase de synchronisation incrémentielle.
Vérifiez la synchronisation des données et des modifications de schéma en temps réel.

La source MySQL CDC prend en charge la synchronisation des données et du schéma en temps réel pendant la phase incrémentielle. Pour le vérifier, modifiez le schéma et les données d'une table utilisateur fragmentée dans MySQL une fois que la tâche est entrée dans cette phase.
1. Connectez-vous à l'instance ApsaraDB RDS for MySQL via DMS.
  
  Pour plus d'informations, consultez (Obsolète, redirection vers « Étape 2 ») Se connecter à une instance ApsaraDB RDS for MySQL via DMS.
2. Dans la base de données user_db2, exécutez les commandes suivantes pour modifier le schéma de la table user02 et insérer et mettre à jour des données.
```
USE `user_db2`;
ALTER TABLE `user02` ADD COLUMN `age` INT;   -- Add the age column.
INSERT INTO `user02` (id, name, age) VALUES (27, 'Tony', 30); -- Insert a row that includes the age data.
UPDATE `user05` SET name='JARK' WHERE id=15;  -- Update another table and change the name to uppercase.
```
3. Dans la console Hologres, vérifiez les modifications apportées au schéma et aux données de la table users.
  Dans le coin supérieur droit de la page d'informations de la table users, cliquez sur Query Table, saisissez la commande suivante, puis cliquez sur Run.
  select * from users order by _db_name,_table_name,id;
  La modification du schéma sur user02 et les modifications de données sont propagées en temps réel, même si les tables fragmentées ont des schémas différents. La table users Hologres affiche désormais la nouvelle colonne age, l'enregistrement inséré pour Tony et l'enregistrement mis à jour pour JARK.
  
  Les résultats de la requête affichent 11 lignes de la table users provenant de user_db1, user_db2 et user_db3. Les modifications de synchronisation incrémentielle sont reflétées : l'utilisateur Tony (id 27) dans user_db2 a age mis à jour à 30, et un autre enregistrement a name changé en JARK. Les lignes restantes ont des valeurs d'âge de \N, confirmant que la synchronisation incrémentielle fonctionne.

(Facultatif) Étape 5 : Configurer les ressources de la tâche

Pour de meilleures performances, vous pouvez ajuster les ressources de la tâche telles que la concurrence, la mémoire TaskManager et les CU en fonction de votre volume de données.

Sur la page O&M > Deployments, cliquez sur le nom de la tâche cible.
Sous l'onglet Configuration, cliquez sur Edit dans le coin supérieur droit de la section Resources.
Définissez manuellement les paramètres de ressource tels que la mémoire TaskManager et la concurrence.
À droite de la section Resources, cliquez sur Save.
Redémarrez la tâche.

Les modifications de configuration des ressources ne prennent effet qu'après le redémarrage de la tâche.

Documents connexes

Pour la syntaxe de chaque module d'ingestion de données, consultez Référence de développement de tâche d'ingestion de données Flink CDC.
Si vous rencontrez des problèmes lors de l'exécution de votre tâche d'ingestion de données, consultez Problèmes courants et solutions pour les tâches d'ingestion de données.
Ingestion de journaux en temps réel dans un entrepôt de données
Mettre en œuvre l'ingestion en temps réel dans un entrepôt de données
Ingestion de base de données en temps réel dans un lac de données
Construire un entrepôt de données en temps réel avec Hologres
Construire un lakehouse en streaming avec Paimon et StarRocks