Data Transmission Service:Configurer ETL dans les tâches de migration ou de synchronisation DTS

DTS permet de configurer ETL de deux manières :

• Configurer ETL lors de la création d'une tâche de synchronisation

• Modifier la configuration ETL d'une tâche de synchronisation existante

Remarques

• Si votre script ETL ajoute de nouvelles colonnes, ajoutez-les manuellement à la base de données de destination. Dans le cas contraire, le script ETL ne prendra pas effet. Par exemple, dans e_set(, vous devez ajouter manuellement new_column à la base de données de destination.

• Les scripts DSL gèrent uniquement les opérations de transformation et de nettoyage des données. Ils ne permettent pas de créer des objets de base de données.

• Les champs référencés dans un script DSL doivent exister dans la base de données source et ne doivent être exclus par aucune condition de filtrage. Sinon, la tâche risque d'échouer.

• Les scripts DSL respectent la casse. Les noms de bases de données, de tables et de champs doivent correspondre exactement à ceux de la base de données source.

• Les scripts DSL ne prennent pas en charge les expressions multiples. Utilisez la fonction e_compose pour combiner plusieurs expressions en une seule.

• Toutes les modifications DML provenant de toutes les tables de la base de données source doivent produire des informations de colonnes identiques après le traitement DSL. Autrement, la tâche peut échouer. Par exemple, si vous utilisez la fonction e_set pour ajouter une colonne, assurez-vous que les opérations INSERT, UPDATE et DELETE de la base de données source aboutissent toutes à la même colonne supplémentaire dans la table de destination. Pour plus d'informations, consultez Enregistrer l'heure de modification des données.

Procédure

1. Créez une tâche de synchronisation. Pour plus de détails, consultez Présentation des solutions de synchronisation.

2. À l'étape Advanced Configurations, définissez Configure ETL sur Yes.

3. Dans la zone de saisie, saisissez votre instruction ETL en utilisant la syntaxe DSL de traitement des données.

   Remarque

   Par exemple, pour supprimer les enregistrements dont l'ID est supérieur à 3, utilisez e_if(op_gt(. Ici, op_gt est une fonction d'expression qui vérifie si une valeur est supérieure à une autre, et id est une variable. Ce script filtre les enregistrements où id > 3.

4. Terminez les étapes restantes selon vos besoins.

La modification de la configuration ETL d'une tâche de synchronisation existante comprend les cas suivants :

• Si vous disposez d'une tâche de synchronisation existante sans configuration ETL (c'est-à-dire que lors de sa création, Configure ETL était défini sur No), vous pouvez passer No à Yes et configurer un script DSL.

• Si ETL était déjà configuré, vous pouvez modifier le script DSL existant ou définir Configure ETL sur No.

  Important

  • Pour modifier un script DSL existant, déplacez d'abord l'objet de synchronisation depuis Selected Objects vers Source Objects, puis rajoutez-le à Selected Objects avant de modifier le script DSL.

  • Les tâches de migration ne prennent pas en charge la modification des scripts DSL.

Remarques

• La modification de la configuration ETL d'une tâche de synchronisation existante ne permet pas de changer le schéma de table dans la base de données de destination. Pour modifier le schéma, effectuez cette opération dans la base de données de destination avant de démarrer la tâche de synchronisation.

• La modification de la configuration ETL peut interrompre la liaison de données. Procédez avec prudence.

• Les changements de configuration ETL s'appliquent uniquement aux données incrémentielles traitées après le redémarrage de la tâche. Ils n'affectent pas les données historiques traitées avant la modification.

• Les scripts DSL gèrent uniquement les opérations de transformation et de nettoyage des données. Ils ne permettent pas de créer des objets de base de données.

• Les champs référencés dans un script DSL doivent exister dans la base de données source et ne doivent être exclus par aucune condition de filtrage. Sinon, la tâche risque d'échouer.

• Les scripts DSL respectent la casse. Les noms de bases de données, de tables et de champs doivent correspondre exactement à ceux de la base de données source.

• Les scripts DSL ne prennent pas en charge les expressions multiples. Utilisez la fonction e_compose pour combiner plusieurs expressions en une seule.

• Toutes les modifications DML provenant de toutes les tables de la base de données source doivent produire des informations de colonnes identiques après le traitement DSL. Autrement, la tâche peut échouer. Par exemple, si vous utilisez la fonction e_set pour ajouter une colonne, assurez-vous que les opérations INSERT, UPDATE et DELETE de la base de données source aboutissent toutes à la même colonne supplémentaire dans la table de destination. Pour plus d'informations, consultez Enregistrer l'heure de modification des données.

Procédure

1. Connectez-vous à la page de liste des nouvelles tâches de synchronisation DTS.

2. Sur la tâche de synchronisation cible, cliquez sur et sélectionnez Modify ETL Configurations.

3. À l'étape Advanced Configurations, définissez Configure ETL sur Yes.

4. Dans la zone de saisie, saisissez votre instruction ETL en utilisant la syntaxe DSL de traitement des données.

   Remarque

   Par exemple, pour supprimer les enregistrements dont l'ID est supérieur à 3, utilisez e_if(op_gt(. Ici, op_gt est une fonction d'expression qui vérifie si une valeur est supérieure à une autre, et id est une variable. Ce script filtre les enregistrements où id > 3.

5. Terminez les étapes restantes selon vos besoins.

Le DSL de traitement des données est un langage de script permettant de définir la logique ETL dans les scénarios de synchronisation DTS. Il prend en charge les fonctions conditionnelles et gère les chaînes, les dates et les valeurs numériques avec les caractéristiques suivantes :

• Fonctionnalités puissantes : offre de nombreuses fonctions et permet la composition de fonctions.

• Syntaxe relativement simple : inclut des exemples pour les scénarios courants tels que le filtrage, la transformation et le masquage des données. Pour plus de détails, consultez Exemples de scénarios typiques.

• Haute efficacité d'exécution : utilise la technologie de génération de code pour minimiser l'impact sur les performances du processus de synchronisation.

Filtrage des données

• Filtrage par colonne numérique : supprimez les enregistrements où id > 10 000 afin qu'ils ne soient pas synchronisés : e_if(op_gt(id, 10000), e_drop()).

• Filtrage par correspondance de chaîne : supprimez les enregistrements où name contient « hangzhou » : e_if(str_contains(name, "hangzhou"), e_drop()).

• Filtrage par date : ne synchronisez pas les enregistrements dont order_timestamp est antérieur à une heure spécifique : e_if(op_lt(order_timestamp, "2015-02-23 23:54:55"), e_drop()).

• Filtrage par conditions multiples :

  • Supprimez les enregistrements où id > 1 000 et name contient « hangzhou » : e_if(op_and(str_contains(name, "hangzhou"), op_gt(id, 1000)), e_drop()).

  • Supprimez les enregistrements où id > 1 000 ou name contient « hangzhou » : e_if(op_or(str_contains(name, "hangzhou"), op_gt(id, 1000)), e_drop()).

Masquage des données

Masquage : remplacez les quatre derniers chiffres de la colonne phone par des astérisques : e_set(phone, str_mask(phone, 7, 10, '*')).

Enregistrer l'heure de modification des données

• Ajouter une colonne à toutes les tables : lorsque __OPERATION__ est INSERT, UPDATE ou DELETE, ajoutez une colonne nommée « dts_sync_time » avec la valeur de l'horodatage de validation du journal (__COMMIT_TIMESTAMP__).

  e_if(op_or(op_or(
          op_eq(__OPERATION__, __OP_INSERT__),
          op_eq(__OPERATION__, __OP_UPDATE__)),
          op_eq(__OPERATION__, __OP_DELETE__)),
      e_set(dts_sync_time, __COMMIT_TIMESTAMP__))

• Ajouter une colonne à une table spécifique « dts_test_table » : lorsque __OPERATION__ est INSERT, UPDATE ou DELETE, ajoutez une colonne nommée « dts_sync_time » avec la valeur de l'horodatage de validation du journal (__COMMIT_TIMESTAMP__).

  e_if(op_and(
        op_eq(__TB__,'dts_test_table'),
        op_or(op_or(
          op_eq(__OPERATION__,__OP_INSERT__),
          op_eq(__OPERATION__,__OP_UPDATE__)),
          op_eq(__OPERATION__,__OP_DELETE__))),
        e_set(dts_sync_time,__COMMIT_TIMESTAMP__))

  Remarque

  Vous devez ajouter manuellement la colonne « dts_sync_time » à la table de destination avant de démarrer la tâche.

Audit des changements de données

Enregistrez le type et l'heure des changements de données : écrivez le type de changement dans la colonne « operation_type » et l'horodatage du changement dans la colonne « updated » de la base de données de destination.

e_compose(
    e_switch(
        op_eq(__OPERATION__,__OP_DELETE__), e_set(operation_type, 'DELETE'),
        op_eq(__OPERATION__,__OP_UPDATE__), e_set(operation_type, 'UPDATE'),
        op_eq(__OPERATION__,__OP_INSERT__), e_set(operation_type, 'INSERT')),
    e_set(updated, __COMMIT_TIMESTAMP__),
    e_set(__OPERATION__,__OP_INSERT__)
)

Distinguer les données complètes et incrémentielles

Enregistrez si les données proviennent d'une migration complète ou incrémentielle dans la colonne is_increment_dml. Vous pouvez faire la distinction entre migration complète et incrémentielle en vérifiant la valeur de __COMMIT_TIMESTAMP__. Lors d'une migration complète, __COMMIT_TIMESTAMP__ vaut 0 (1970-01-01 08:00:00, selon le fuseau horaire). Lors d'une migration incrémentielle, cette valeur reflète l'heure d'écriture du journal de la base de données source. Le script ETL correspondant est le suivant :

e_if_else(__COMMIT_TIMESTAMP__ > DATETIME('2000-01-01 00:00:00'), 
    e_set(`is_increment_dml`, True),
    e_set(`is_increment_dml`, False)
)

Constantes et variables

• Constantes

  Type	Exemple
  int	123
  float	123,4
  string	"hello1_world"
  boolean	true or false
  datetime	DATETIME('2021-01-01 10:10:01')

• Variables

  Variable	Description	Type de données	Valeur d'exemple
  __TB__	Nom de la table	string	table
  __DB__	Nom de la base de données	string	mydb
  __OPERATION__	Type d'opération	string	__OP_INSERT__, __OP_UPDATE__, __OP_DELETE__
  __BEFORE__	Valeur pré-image pour les opérations UPDATE (valeur avant modification) Remarque Les opérations DELETE ne comportent que des valeurs pré-image.	Marqueur spécial, sans type	v(column_name,__BEFORE__)
  __AFTER__	Valeur post-image pour les opérations UPDATE (valeur après modification) Remarque Les opérations INSERT ne comportent que des valeurs post-image.	Marqueur spécial, sans type	v(column_name,__AFTER__)
  __COMMIT_TIMESTAMP__	Heure de validation de la transaction	datetime	'2021-01-01 10:10:01'
  column	Valeur de la colonne spécifiée pour un enregistrement	string	id, name

Fonctions d'expression

• Opérations numériques

  Fonction	Syntaxe	Plage de valeurs	Valeur de retour	Exemple
  Addition	op_sum(value1, value2)	value1 : nombre entier ou à virgule flottante value2 : nombre entier ou à virgule flottante	Renvoie un entier si les deux paramètres sont des entiers ; sinon, renvoie un nombre à virgule flottante.	op_sum(col1, 1,0)
  Soustraction	op_sub(value1, value2)	value1 : nombre entier ou à virgule flottante value2 : nombre entier ou à virgule flottante	Renvoie un entier si les deux paramètres sont des entiers ; sinon, renvoie un nombre à virgule flottante.	op_sub(col1, 1,0)
  Multiplication	op_mul(value1, value2)	value1 : nombre entier ou à virgule flottante value2 : nombre entier ou à virgule flottante	Renvoie un entier si les deux paramètres sont des entiers ; sinon, renvoie un nombre à virgule flottante.	op_mul(col1, 1,0)
  Division	op_div_true(value1, value2)	value1 : nombre entier ou à virgule flottante value2 : nombre entier ou à virgule flottante	Renvoie un entier si les deux paramètres sont des entiers ; sinon, renvoie un nombre à virgule flottante.	op_div_true(col1, 2,0) ; si col1=15, renvoie 7,5.
  Opération modulo	op_mod(value1, value2)	value1 : nombre entier ou à virgule flottante value2 : nombre entier ou à virgule flottante	Renvoie un entier si les deux paramètres sont des entiers ; sinon, renvoie un nombre à virgule flottante.	op_mod(col1, 10) ; si col1=23, renvoie 3.

• Opérations logiques

  Fonction	Syntaxe	Valeurs valides	Valeur de retour	Exemple
  Vérification d'égalité	op_eq(value1, value2)	value1 : entier, nombre à virgule flottante ou chaîne value2 : entier, nombre à virgule flottante ou chaîne	Booléen : true ou false	op_eq(col1, 23)
  Vérification de supériorité	op_gt(value1, value2)	value1 : entier, nombre à virgule flottante ou chaîne value2 : entier, nombre à virgule flottante ou chaîne	Booléen : true ou false	op_gt(col1, 1,0)
  Est inférieur à	op_lt(value1, value2)	value1 : entier, nombre à virgule flottante ou chaîne value2 : entier, nombre à virgule flottante ou chaîne	Booléen : true ou false	op_lt(col1, 1,0)
  Vérification de supériorité ou d'égalité	op_ge(value1, value2)	value1 : entier, nombre à virgule flottante ou chaîne value2 : entier, nombre à virgule flottante ou chaîne	Booléen : true ou false	op_ge(col1, 1,0)
  Vérification d'infériorité ou d'égalité	op_le(value1, value2)	value1 : entier, nombre à virgule flottante ou chaîne value2 : entier, nombre à virgule flottante ou chaîne	Booléen : true ou false	op_le(col1, 1,0)
  Opération ET	op_and(value1, value2)	value1 : booléen value2 : booléen	Booléen : true ou false	op_and(is_male, is_student)
  Opération OU	op_or(value1, value2)	value1 : booléen value2 : booléen	Booléen : true ou false	op_or(is_male, is_student)
  Opération IN	op_in(value, json_array)	value : tout type json_array : chaîne au format JSON	Booléen : true ou false	op_in(id,json_array('["0","1","2","3","4","5","6","7","8"]'))
  Vérifier si la valeur est nulle	op_is_null(value)	value : tout type	Booléen : true ou false	op_is_null(name)
  Vérifier si la valeur n'est pas nulle	op_is_not_null(value)	value : tout type	Booléen : true ou false	op_is_not_null(name)

• Fonctions de chaîne

  Fonction	Syntaxe	Plage de valeurs	Valeur de retour	Exemple
  Concaténation de chaînes	op_add(str_1,str_2,...,str_n)	str_1 : chaîne str_2 : chaîne ... str_n : chaîne	Chaîne concaténée	op_add(col,'hangzhou','dts')
  Formatage et concaténation de chaînes	str_format(format, value1, value2, value3, ...)	format : chaîne avec des accolades comme espaces réservés, par exemple "part1: {}, part2: {}" value1 : tout type value2 : tout type	Chaîne formatée	str_format("part1: {}, part2: {}", col1, col2) ; si col1="ab" et col2="12", renvoie "part1: ab, part2: 12".
  Remplacement de chaîne	str_replace(original, oldStr, newStr, count)	original : chaîne d'origine oldStr : chaîne à remplacer newStr : chaîne de remplacement count : entier, nombre maximal de remplacements. Définissez -1 pour toutes les occurrences.	Chaîne remplacée	str_replace(name, "a", 'b', 1) ; si name="aba", renvoie "bba". str_replace(name, "a", 'b', -1) ; si name="aba", renvoie "bbb".
  Remplacer les valeurs dans tous les champs de type chaîne (tels que varchar, text, char)	tail_replace_string_field(search, replace, all)	search : chaîne à remplacer replace : chaîne de remplacement all : indique s'il faut remplacer toutes les correspondances ; prend actuellement en charge uniquement true. Remarque Si vous ne souhaitez pas remplacer toutes les correspondances, utilisez la fonction str_replace.	Chaîne remplacée	tail_replace_string_field('\u000f', '', true) remplace toutes les occurrences de "\u000f" dans les valeurs des champs de type chaîne par un espace.
  Supprimer les caractères spécifiés au début et à la fin de la chaîne	str_strip(string_val, charSet)	string_val : chaîne d'origine char_set : ensemble de caractères à supprimer	Chaîne avec les caractères de début et de fin supprimés	str_strip(name, 'ab') ; si name=axbzb, renvoie xbz.
  Convertir la chaîne en minuscules	str_lower(value)	value : colonne de chaîne ou littéral de chaîne	Chaîne en minuscules	str_lower(str_col)
  Convertir la chaîne en majuscules	str_upper(value)	value : colonne de chaîne ou littéral de chaîne	Chaîne en majuscules	str_upper(str_col)
  Convertir une chaîne en nombre	cast_string_to_long(value)	value : chaîne	Entier	cast_string_to_long(col)
  Convertir un nombre en chaîne	cast_long_to_string(value)	value : entier	Chaîne	cast_long_to_string(col)
  Compter les occurrences d'une sous-chaîne	str_count(str,pattern)	str : colonne de chaîne ou littéral de chaîne pattern : sous-chaîne à rechercher	Nombre d'apparitions de la sous-chaîne	str_count(str_col, 'abc') ; si str_col="zabcyabcz", renvoie 2.
  Trouver la position d'une sous-chaîne	str_find(str, pattern)	str : colonne de chaîne ou littéral de chaîne pattern : sous-chaîne à rechercher	Position de la première correspondance ; renvoie -1 si aucune correspondance n'est trouvée	str_find(str_col, 'abc') ; si str_col="xabcy", renvoie 1.
  Vérifier si la chaîne contient uniquement des lettres	str_isalpha(str)	str : colonne de chaîne ou littéral de chaîne	true ou false	str_isalpha(str_col)
  Vérifier si la chaîne contient uniquement des chiffres	str_isdigit(str)	str : colonne de chaîne ou littéral de chaîne	true ou false	str_isdigit(str_col)
  Correspondance d'expression régulière	regex_match(str,regex)	str : colonne de chaîne ou littéral de chaîne regex : colonne de chaîne d'expression régulière ou littéral de chaîne	true ou false	regex_match(__TB__,'user_\\d+')
  Masquer une partie d'une chaîne avec un caractère spécifié pour l'anonymisation des données, par exemple remplacer les quatre derniers chiffres d'un numéro de téléphone par des astérisques	str_mask(str, start, end, maskStr)	str : colonne de chaîne ou littéral de chaîne start : entier, position de départ du masquage (minimum 0) end : entier, position de fin du masquage (longueur maximale de la chaîne moins un) maskStr : chaîne à caractère unique, telle que '#'	Chaîne avec les caractères masqués de start à end	str_mask(phone, 7, 10, '#')
  Extraire la partie de la chaîne après cond	substring_after(str, cond)	str : chaîne d'origine cond : chaîne	Chaîne Remarque Le résultat n'inclut pas cond.	substring_after(col, 'abc')
  Extraire la partie de la chaîne avant cond	substring_before(str, cond)	str : chaîne d'origine cond : chaîne	Chaîne Remarque Le résultat n'inclut pas cond.	substring_before(col, 'efg')
  Extraire la partie de la chaîne entre cond1 et cond2	substring_between(str, cond1, cond2)	str : chaîne d'origine cond1 : chaîne cond2 : chaîne	Chaîne Remarque Le résultat n'inclut ni cond1 ni cond2.	substring_between(col, 'abc','efg')
  Vérifier si la valeur est de type chaîne	is_string_value(value)	value : chaîne ou nom de colonne	Booléen : true ou false	is_string_value(col1)
  Remplacer le contenu dans les champs de type chaîne ; commence par la fin dans l'ordre inverse	tail_replace_string_field(search, replace, all)	search : chaîne à remplacer replace : chaîne de remplacement all : indique s'il faut tout remplacer ; true ou false	Chaîne remplacée	Remplace "\u000f" par un espace dans toutes les valeurs de champ de chaîne. tail_replace_string_field('\u000f','',true)
  Obtenir la valeur d'un champ dans MongoDB	bson_value("field1","field2","field3",...)	field1 : nom de champ de premier niveau field2 : nom de champ de deuxième niveau	Valeur du champ spécifié dans le document	e_set(user_id, bson_value("id")) e_set(user_name, bson_value("person","name"))

Fonctions globales

• Fonctions de contrôle de flux

  Fonction	Syntaxe	Paramètres	Exemple
  Instruction if	e_if(bool_expr, func_invoke)	bool_expr : constante booléenne ou appel de fonction. Constantes : true ou false. Exemple d'appel de fonction : op_gt(id, 10). func_invoke : appel de fonction. Fonctions prises en charge : e_drop, e_keep, e_set, e_if, e_compose.	e_if(op_gt(id, 10), e_drop()) ; supprime l'enregistrement si l'ID > 10.
  Instruction if-else	e_if_else(bool_expr, func_invoke1, func_invoke2)	bool_expr : constante booléenne ou appel de fonction. Constantes : true ou false. Exemple d'appel de fonction : op_gt(id, 10). func_invoke1 : appel de fonction exécuté si la condition est vraie. func_invoke2 : appel de fonction exécuté si la condition est fausse.	e_if_else(op_gt(id, 10), e_set(tag, 'large'), e_set(tag, 'small')) ; définit le tag à « large » si l'ID > 10, sinon à « small ».
  Instruction de type switch évaluant plusieurs conditions et exécutant la première opération correspondante. Une opération par défaut s'exécute si aucune condition ne correspond.	s_switch(condition1, func1, condition2, func2, ..., default=default_func)	condition1 : constante booléenne ou appel de fonction. Constantes : true ou false. Exemple d'appel de fonction : op_gt(id, 10). func_invoke : appel de fonction. Évalue condition1 ; si vraie, exécute cette fonction et quitte le switch. Si fausse, passe à la condition suivante. default_func : appel de fonction exécuté si toutes les conditions sont fausses.	e_switch(op_gt(id, 100), e_set(str_col, '>100'), op_gt(id, 90), e_set(str_col, '>90'), default=e_set(str_col, '<=90')).
  Combinaison de plusieurs opérations	e_compose(func1, func2, func3, ...)	func1 : appel de fonction. Peut être e_set, e_drop ou e_if. func2 : appel de fonction. Peut être e_set, e_drop ou e_if.	e_compose(e_set(str_col, 'test'), e_set(dt_col, dt_now())) ; définit str_col à « test » et dt_col à l'heure actuelle.

• Fonctions de manipulation des données

  Fonction	Syntaxe	Paramètres	Exemple
  Supprimer cet enregistrement (ne pas synchroniser)	e_drop()	Aucun	e_if(op_gt(id, 10), e_drop()) ; supprime les enregistrements dont l'ID > 10.
  Conserver cet enregistrement (synchroniser vers la destination)	e_keep(condition)	condition : expression booléenne	e_keep(op_gt(id, 1)) ; synchronise uniquement les enregistrements dont l'ID > 1.
  Définir la valeur d'une colonne	e_set(col, val, NEW)	col : nom de la colonne val : constante ou appel de fonction. Doit correspondre au type de données de col. NEW : convertit col au type de données de val (facultatif) Important Si vous omettez NEW, n'incluez pas la virgule précédente. Assurez-vous de la compatibilité des types de données pour éviter des erreurs de tâche.	e_set(dt_col, dt_now()) ; définit dt_col à l'heure actuelle. e_set(col1, col2 + 1) ; définit col1 à col2 + 1. e_set(col1, 1, NEW) ; convertit col1 en type numérique et lui attribue la valeur 1.
  Conservation, suppression et mappage des noms de champs MongoDB	e_expand_bson_value('*', 'fieldA',{"fieldB":"fieldC"})	: noms de champs à conserver ; signifie tous les champs. fieldA : noms de champs à supprimer. {"fieldB":"fieldC"} : mappage des noms de champs ; fieldB est le nom du champ source, fieldC est le nom du champ de destination. Remarque Le mappage des noms de champs est facultatif.	e_expand_bson_value("*", "_id,name") ; écrit tous les champs sauf _id et name vers la destination.