Data Management:Orchestration des tâches

Bases de données prises en charge

Task Orchestration prend en charge les types de bases de données suivants :

Bases de données relationnelles

• MySQL : ApsaraDB RDS for MySQL, PolarDB for MySQL, ApsaraDB MyBase for MySQL, PolarDB for Xscale et bases de données MySQL provenant d'autres sources

• SQL Server : ApsaraDB RDS for SQL Server, ApsaraDB MyBase for SQL Server et bases de données SQL Server provenant d'autres sources

• PostgreSQL : ApsaraDB RDS for PostgreSQL, PolarDB for PostgreSQL, ApsaraDB MyBase for PostgreSQL et bases de données PostgreSQL provenant d'autres sources

• OceanBase : ApsaraDB for OceanBase en mode MySQL, ApsaraDB for OceanBase en mode Oracle et bases de données OceanBase gérées par l'utilisateur

• PolarDB for PostgreSQL (compatible avec Oracle)

• Oracle

• Dameng (DM)

• Db2

Base de données NoSQL : Lindorm

Entrepôts de données : AnalyticDB for MySQL, AnalyticDB for PostgreSQL, Data Lake Analytics (DLA), MaxCompute, Hologres

Stockage objet : Object Storage Service (OSS)

-- Create the source table
CREATE TABLE test (
    `id` bigint(20) UNSIGNED NOT NULL PRIMARY KEY AUTO_INCREMENT COMMENT 'Primary key',
    `gmt_create` DATETIME NOT NULL COMMENT 'Creation time',
    `gmt_modified` DATETIME NOT NULL COMMENT 'Modification time',
    `content` TEXT COMMENT 'Test data'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='Table for a task orchestration test';

-- Insert sample data
INSERT INTO test(`gmt_create`, `gmt_modified`, `content`) VALUES ('2020-01-01 01:00:00', '2020-01-01 01:00:00', 'value1');
INSERT INTO test(`gmt_create`, `gmt_modified`, `content`) VALUES ('2020-02-01 01:00:00', '2020-02-01 01:00:00', 'value2');
INSERT INTO test(`gmt_create`, `gmt_modified`, `content`) VALUES ('2020-03-01 01:00:00', '2020-03-01 01:00:00', 'value3');
INSERT INTO test(`gmt_create`, `gmt_modified`, `content`) VALUES ('2020-04-01 01:00:00', '2020-04-01 01:00:00', 'value4');
INSERT INTO test(`gmt_create`, `gmt_modified`, `content`) VALUES ('2020-05-01 01:00:00', '2020-05-01 01:00:00', 'value5');
INSERT INTO test(`gmt_create`, `gmt_modified`, `content`) VALUES ('2020-06-01 01:00:00', '2020-06-01 01:00:00', 'value6');
INSERT INTO test(`gmt_create`, `gmt_modified`, `content`) VALUES ('2020-07-01 01:00:00', '2020-07-01 01:00:00', 'value7');
INSERT INTO test(`gmt_create`, `gmt_modified`, `content`) VALUES ('2020-08-01 01:00:00', '2020-08-01 01:00:00', 'value8');
INSERT INTO test(`gmt_create`, `gmt_modified`, `content`) VALUES ('2020-09-01 01:00:00', '2020-09-01 01:00:00', 'value9');
INSERT INTO test(`gmt_create`, `gmt_modified`, `content`) VALUES ('2020-10-01 01:00:00', '2020-10-01 01:00:00', 'value10');

Créer le nœud Migrate Data

Ce nœud crée une table d'archivage mensuelle (si elle n'existe pas déjà) et y copie tous les enregistrements datant de plus de six mois depuis la table test.

1. Depuis le panneau Task Type situé à gauche du canevas, faites glisser un nœud Single Instance SQL vers le canevas.

2. Cliquez avec le bouton droit sur le nœud et sélectionnez Rename. Saisissez Migrate Data comme nom de nœud.

3. Double-cliquez sur le nœud pour ouvrir l'éditeur SQL.

4. Sélectionnez votre base de données dans la zone de recherche de bases de données et saisissez le code SQL suivant :

   CREATE TABLE IF NOT EXISTS `test_${yearmonth6_name}` LIKE test;
   INSERT INTO `test_${yearmonth6_name}`
   SELECT * FROM `test`
   WHERE gmt_create < '${yearmonth6_date}';

   La première instruction crée la table test_${yearmonth6_name} avec le même schéma que la table test, en ignorant la création si la table existe déjà. La seconde instruction copie toutes les lignes dont la date gmt_create est antérieure à la limite de six mois dans la table d'archivage.

5. Cliquez sur Preview pour valider les instructions SQL.

Créer le nœud Clear Historical Data

Ce nœud supprime de la table test toutes les lignes qui ont été copiées dans la table d'archivage.

1. Faites glisser un autre nœud Single Instance SQL vers le canevas.

2. Cliquez avec le bouton droit sur le nœud et sélectionnez Rename. Saisissez Clear Historical Data comme nom de nœud.

3. Double-cliquez sur le nœud pour ouvrir l'éditeur SQL.

4. Sélectionnez votre base de données dans la zone de recherche de bases de données et saisissez le code SQL suivant :

   Avertissement

   Exécutez l'instruction DELETE pendant les heures creuses. La suppression de grands volumes de données peut maintenir des verrous de table pendant une période prolongée et interrompre d'autres opérations.

   DELETE FROM `test` WHERE gmt_create < '${yearmonth6_date}';

   Cette instruction supprime toutes les lignes déjà copiées dans la table d'archivage, garantissant ainsi que la table test ne contient aucun enregistrement datant de plus de six mois.

5. Cliquez sur Preview pour valider l'instruction SQL.

Connecter les nœuds

Sur le canevas, survolez le nœud Migrate Data, cliquez sur le cercle vide situé à sa droite et faites glisser la ligne de connexion vers le nœud Clear Historical Data. Cela définit l'ordre d'exécution : Migrate Data s'exécute en premier, puis Clear Historical Data.

Configurer la planification

1. En bas de la page, cliquez sur l'onglet Task Flow Information.

2. Dans la section Scheduling Settings, configurez les paramètres suivants :

   Paramètre	Valeur
   Enable scheduling	Activer
   Scheduling type	Timed Scheduling/Cyclic scheduling
   Effective time	1970-01-01 – 9999-01-01 (par défaut ; la planification s'exécute indéfiniment)
   Scheduling cycle	Month
   Specified time	1st of each month
   Specific point in time	01:00
   Time zone	Votre fuseau horaire préféré
   Cron expression	Généré automatiquement en fonction des paramètres ci-dessus

   Cet exemple planifie l'exécution du flux de tâches à 01:00 le premier jour de chaque mois. Ajustez les paramètres selon vos besoins. Pour plus d'options, consultez la rubrique Overview.

Test d'exécution

Cliquez sur Try Run dans le coin supérieur gauche du canevas. Vérifiez la sortie du journal :

• status SUCCEEDED sur la dernière ligne : le test d'exécution s'est terminé avec succès.

• status FAILED sur la dernière ligne : le test d'exécution a échoué. Consultez les journaux pour identifier quel nœud a échoué et pourquoi, corrigez la configuration et réessayez.

Vérifier les résultats

Dans l'onglet SQLConsole, exécutez des requêtes pour confirmer que :

• Les enregistrements dont la valeur gmt_create est antérieure à la limite de six mois ont été copiés dans la table d'archivage test_${yearmonth6_name}.

• Ces mêmes enregistrements n'existent plus dans la table test.

Pour plus de détails sur l'interrogation des données, consultez la rubrique Gérer une base de données dans l'onglet SQLConsole.