Data Transmission Service:Configurar ETL em tarefas de migração ou sincronização do DTS

O DTS oferece suporte à configuração de ETL de duas maneiras:

• Configure ETL when creating a synchronization task

• Modify ETL configuration on an existing synchronization task

Notes

• Se o seu script ETL adicionar novas colunas, adicione-as manualmente ao banco de dados de destino. Caso contrário, o script ETL não terá efeito. Por exemplo, em e_set(, você deve adicionar manualmente new_column ao banco de dados de destino.

• Os scripts DSL lidam apenas com operações de transformação e limpeza de dados. Eles não oferecem suporte à criação de objetos de banco de dados.

• Os campos referenciados em um script DSL devem existir no banco de dados de origem e não devem ser filtrados por nenhuma condição de filtro. Caso contrário, a tarefa poderá falhar.

• Os scripts DSL diferenciam maiúsculas de minúsculas. Os nomes de bancos de dados, tabelas e campos devem corresponder exatamente aos do banco de dados de origem.

• Os scripts DSL não oferecem suporte a múltiplas expressões. Use a função e_compose para combinar várias expressões em uma só.

• Todas as alterações DML de todas as tabelas no banco de dados de origem devem produzir informações de coluna idênticas após o processamento DSL. Caso contrário, a tarefa poderá falhar. Por exemplo, se você usar a função e_set para adicionar uma coluna, certifique-se de que as operações INSERT, UPDATE e DELETE do banco de dados de origem resultem todas na mesma coluna adicional na tabela de destino. Para mais informações, consulte Record data modification time.

Procedure

1. Crie uma tarefa de sincronização. Para obter detalhes, consulte Overview of synchronization solutions.

2. Na etapa Advanced Configurations, defina Configure ETL como Yes.

3. Na caixa de entrada, insira sua instrução ETL usando a data processing DSL syntax.

   Nota

   Por exemplo, para descartar registros onde id é maior que 3, use e_if(op_gt(. Aqui, op_gt é uma função de expressão que verifica se um valor é maior que outro, e id é uma variável. Este script filtra os registros onde id > 3.

4. Conclua as etapas restantes conforme necessário.

A modificação da configuração de ETL em uma tarefa de sincronização existente inclui:

• Se você tiver uma tarefa de sincronização existente que não tenha o ETL configurado — ou seja, quando você criou a tarefa de sincronização, Configure ETL foi definido como No — você pode alterar No para Yes e configurar um script DSL.

• Se o ETL já estava configurado, você pode modificar o script DSL existente ou definir Configure ETL como No.

  Importante

  • Para modificar um script DSL existente, primeiro mova o objeto de sincronização de Selected Objects de volta para Source Objects, e então readicione-o a Selected Objects antes de editar o script DSL.

  • As tarefas de migração não oferecem suporte à modificação de scripts DSL.

Notes

• A modificação da configuração de ETL em uma tarefa de sincronização existente não oferece suporte à alteração do schema da tabela no banco de dados de destino. Para alterar o schema, faça-o no banco de dados de destino antes de iniciar a tarefa de sincronização.

• A modificação da configuração de ETL pode interromper o link de dados. Prossiga com cautela.

• As alterações na configuração de ETL aplicam-se apenas aos dados incrementais processados após a reinicialização da tarefa. Elas não afetam os dados históricos processados antes da alteração.

• Os scripts DSL lidam apenas com operações de transformação e limpeza de dados. Eles não oferecem suporte à criação de objetos de banco de dados.

• Os campos referenciados em um script DSL devem existir no banco de dados de origem e não devem ser filtrados por nenhuma condição de filtro. Caso contrário, a tarefa poderá falhar.

• Os scripts DSL diferenciam maiúsculas de minúsculas. Os nomes de bancos de dados, tabelas e campos devem corresponder exatamente aos do banco de dados de origem.

• Os scripts DSL não oferecem suporte a múltiplas expressões. Use a função e_compose para combinar várias expressões em uma só.

• Todas as alterações DML de todas as tabelas no banco de dados de origem devem produzir informações de coluna idênticas após o processamento DSL. Caso contrário, a tarefa poderá falhar. Por exemplo, se você usar a função e_set para adicionar uma coluna, certifique-se de que as operações INSERT, UPDATE e DELETE do banco de dados de origem resultem todas na mesma coluna adicional na tabela de destino. Para mais informações, consulte Record data modification time.

Procedure

1. Faça login no list page of new DTS synchronization tasks.

2. Na tarefa de sincronização de destino, clique em e selecione Modify ETL Configurations.

3. Na etapa Advanced Configurations, defina Configure ETL como Yes.

4. Na caixa de entrada, insira sua instrução ETL usando a data processing DSL syntax.

   Nota

   Por exemplo, para descartar registros onde id é maior que 3, use e_if(op_gt(. Aqui, op_gt é uma função de expressão que verifica se um valor é maior que outro, e id é uma variável. Este script filtra os registros onde id > 3.

5. Conclua as etapas restantes conforme necessário.

A DSL de processamento de dados é uma linguagem de script para definir a lógica de ETL em cenários de sincronização do DTS. Ela oferece suporte a funções condicionais e lida com strings, datas e valores numéricos com os seguintes recursos:

• Funcionalidade poderosa: Oferece muitas funções e suporta composição de funções.

• Sintaxe relativamente simples: Inclui exemplos para cenários comuns, como filtragem, transformação e mascaramento de dados. Para obter detalhes, consulte Typical scenario examples.

• Alta eficiência de execução: Usa tecnologia de geração de código para minimizar o impacto no desempenho do processo de sincronização.

Data filtering

• Filtrar por coluna numérica: Descartar registros onde id > 10000 para que não sejam sincronizados: e_if(op_gt(id, 10000), e_drop()).

• Filtrar por correspondência de string: Descartar registros onde name contém "hangzhou": e_if(str_contains(name, "hangzhou"), e_drop()).

• Filtrar por data: Não sincronizar registros onde order_timestamp é anterior a um horário específico: e_if(op_lt(order_timestamp, "2015-02-23 23:54:55"), e_drop()).

• Filtrar por múltiplas condições:

  • Descartar registros onde id > 1000 e name contém "hangzhou": e_if(op_and(str_contains(name, "hangzhou"), op_gt(id, 1000)), e_drop()).

  • Descartar registros onde id > 1000 ou name contém "hangzhou": e_if(op_or(str_contains(name, "hangzhou"), op_gt(id, 1000)), e_drop()).

Data masking

Mascaramento: Substituir os últimos quatro dígitos da coluna de número de telefone por asteriscos: e_set(phone, str_mask(phone, 7, 10, '*')).

Record data modification time

• Adicionar uma coluna a todas as tabelas: Quando __OPERATION__ for INSERT, UPDATE ou DELETE, adicione uma coluna chamada "dts_sync_time" com o valor do carimbo de data/hora de commit do log (__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__))

• Adicionar uma coluna a uma tabela específica "dts_test_table": Quando __OPERATION__ for INSERT, UPDATE ou DELETE, adicione uma coluna chamada "dts_sync_time" com o valor do carimbo de data/hora de commit do log (__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__))

  Nota

  Você deve adicionar manualmente a coluna "dts_sync_time" à tabela de destino antes de iniciar a tarefa.

Data change auditing

Registrar o tipo e a hora das alterações de dados: Grave o tipo de alteração na coluna "operation_type" e o carimbo de data/hora da alteração na coluna "updated" no banco de dados de destino.

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__)
)

Distinguish between full and incremental data

Registrar se os dados provêm de migração completa ou incremental na coluna is_increment_dml. Você pode distinguir entre migração completa e incremental verificando o valor de __COMMIT_TIMESTAMP__. Na migração completa, __COMMIT_TIMESTAMP__ é 0 (1970-01-01 08:00:00, afetado pelo fuso horário). Na migração incremental, reflete o horário de gravação do log do banco de dados de origem. O script ETL correspondente é:

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

Constants and variables

• Constants

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

• Variables

  Variable	Description	Data type	Example value
  __TB__	Table name	string	table
  __DB__	Database name	string	mydb
  __OPERATION__	Operation type	string	__OP_INSERT__, __OP_UPDATE__, __OP_DELETE__
  __BEFORE__	Pre-image value for UPDATE operations (value before change) Nota DELETE operations have only pre-image values.	Special marker, no type	v(column_name,__BEFORE__)
  __AFTER__	Post-image value for UPDATE operations (value after change) Nota INSERT operations have only post-image values.	Special marker, no type	v(column_name,__AFTER__)
  __COMMIT_TIMESTAMP__	Transaction commit time	datetime	'2021-01-01 10:10:01'
  column	Value of the specified column for a record	string	id, name

Expression functions

• Numeric operations

  Function	Syntax	Value range	Return value	Example
  Addition	op_sum(value1, value2)	value1: integer or floating-point number value2: integer or floating-point number	Returns an integer if both parameters are integers; otherwise, returns a floating-point number.	op_sum(col1, 1.0)
  Subtraction	op_sub(value1, value2)	value1: integer or floating-point number value2: integer or floating-point number	Returns an integer if both parameters are integers; otherwise, returns a floating-point number.	op_sub(col1, 1.0)
  Multiplication	op_mul(value1, value2)	value1: integer or floating-point number value2: integer or floating-point number	Returns an integer if both parameters are integers; otherwise, returns a floating-point number.	op_mul(col1, 1.0)
  Division	op_div_true(value1, value2)	value1: integer or floating-point number value2: integer or floating-point number	Returns an integer if both parameters are integers; otherwise, returns a floating-point number.	op_div_true(col1, 2.0); if col1=15, returns 7.5.
  Modulo operation	op_mod(value1, value2)	value1: integer or floating-point number value2: integer or floating-point number	Returns an integer if both parameters are integers; otherwise, returns a floating-point number.	op_mod(col1, 10); if col1=23, returns 3.

• Logical operations

  Function	Syntax	Valid values	Return value	Example
  Equality check	op_eq(value1, value2)	value1: integer, floating-point number, or string value2: integer, floating-point number, or string	Boolean: true or false	op_eq(col1, 23)
  Greater-than check	op_gt(value1, value2)	value1: integer, floating-point number, or string value2: integer, floating-point number, or string	Boolean: true or false	op_gt(col1, 1.0)
  is less than	op_lt(value1, value2)	value1: integer, floating-point number, or string value2: integer, floating-point number, or string	Boolean: true or false	op_lt(col1, 1.0)
  Greater-than-or-equal check	op_ge(value1, value2)	value1: integer, floating-point number, or string value2: integer, floating-point number, or string	Boolean: true or false	op_ge(col1, 1.0)
  Less-than-or-equal check	op_le(value1, value2)	value1: integer, floating-point number, or string value2: integer, floating-point number, or string	Boolean: true or false	op_le(col1, 1.0)
  AND operation	op_and(value1, value2)	value1: boolean value2: boolean	Boolean: true or false	op_and(is_male, is_student)
  OR operation	op_or(value1, value2)	value1: boolean value2: boolean	Boolean: true or false	op_or(is_male, is_student)
  IN operation	op_in(value, json_array)	value: any type json_array: JSON-formatted string	Boolean: true or false	op_in(id,json_array('["0","1","2","3","4","5","6","7","8"]'))
  Check if value is null	op_is_null(value)	value: any type	Boolean: true or false	op_is_null(name)
  Check if value is not null	op_is_not_null(value)	value: any type	Boolean: true or false	op_is_not_null(name)

• String functions

  Function	Syntax	Value range	Return value	Example
  String concatenation	op_add(str_1,str_2,...,str_n)	str_1: string str_2: string ... str_n: string	Concatenated string	op_add(col,'hangzhou','dts')
  String formatting and concatenation	str_format(format, value1, value2, value3, ...)	format: string with braces as placeholders, such as "part1: {}, part2: {}" value1: any type value2: any type	Formatted string	str_format("part1: {}, part2: {}", col1, col2); if col1="ab" and col2="12", returns "part1: ab, part2: 12".
  String replacement	str_replace(original, oldStr, newStr, count)	original: original string oldStr: string to replace newStr: replacement string count: integer, maximum number of replacements. Set to -1 for all occurrences.	Replaced string	str_replace(name, "a", 'b', 1); if name="aba", returns "bba". str_replace(name, "a", 'b', -1); if name="aba", returns "bbb".
  Replace values in all string-type fields (such as varchar, text, char)	tail_replace_string_field(search, replace, all)	search: string to replace replace: replacement string all: whether to replace all matches; currently supports only true. Nota If you do not want to replace all matches, use the str_replace function.	Replaced string	tail_replace_string_field('\u000f', '', true) replaces all occurrences of "\u000f" in string-type field values with a space.
  Remove specified characters from start and end of string	str_strip(string_val, charSet)	string_val: original string char_set: set of characters to remove	String with leading and trailing characters removed	str_strip(name, 'ab'); if name=axbzb, returns xbz.
  Convert string to lowercase	str_lower(value)	value: string column or string literal	Lowercase string	str_lower(str_col)
  Convert string to uppercase	str_upper(value)	value: string column or string literal	Uppercase string	str_upper(str_col)
  Convert string to number	cast_string_to_long(value)	value: string	Integer	cast_string_to_long(col)
  Convert number to string	cast_long_to_string(value)	value: integer	String	cast_long_to_string(col)
  Count substring occurrences	str_count(str,pattern)	str: string column or string literal pattern: substring to find	Number of times the substring appears	str_count(str_col, 'abc'); if str_col="zabcyabcz", returns 2.
  Find substring position	str_find(str, pattern)	str: string column or string literal pattern: substring to find	Position of first match; returns -1 if not found	str_find(str_col, 'abc'); if str_col="xabcy", returns 1.
  Check if string consists only of letters	str_isalpha(str)	str: string column or string literal	true or false	str_isalpha(str_col)
  Check if string consists only of digits	str_isdigit(str)	str: string column or string literal	true or false	str_isdigit(str_col)
  Regular expression matching	regex_match(str,regex)	str: string column or string literal regex: regular expression string column or string literal	true or false	regex_match(__TB__,'user_\\d+')
  Mask part of a string with a specified character for data masking, such as replacing the last four digits of a phone number with asterisks	str_mask(str, start, end, maskStr)	str: string column or string literal start: integer, starting position for masking (minimum 0) end: integer, ending position for masking (maximum string length minus one) maskStr: single-character string, such as '#'	String with characters from start to end masked	str_mask(phone, 7, 10, '#')
  Extract part of string after cond	substring_after(str, cond)	str: original string cond: string	String Nota Result does not include cond.	substring_after(col, 'abc')
  Extract part of string before cond	substring_before(str, cond)	str: original string cond: string	String Nota Result does not include cond.	substring_before(col, 'efg')
  Extract part of string between cond1 and cond2	substring_between(str, cond1, cond2)	str: original string cond1: string cond2: string	String Nota Result does not include cond1 or cond2.	substring_between(col, 'abc','efg')
  Check if value is a string type	is_string_value(value)	value: string or column name	Boolean: true or false	is_string_value(col1)
  Replace content in string-type fields; starts from the end in reverse order	tail_replace_string_field(search, replace, all)	search: string to replace replace: replacement string all: whether to replace all; true or false	Replaced string	Replace "\u000f" with a space in all string field values. tail_replace_string_field('\u000f','',true)
  Get value of a field in MongoDB	bson_value("field1","field2","field3",...)	field1: top-level field name field2: second-level field name	Value of the specified field in the document	e_set(user_id, bson_value("id")) e_set(user_name, bson_value("person","name"))

Global functions

• Flow control functions

  Function	Syntax	Parameters	Example
  if statement	e_if(bool_expr, func_invoke)	bool_expr: boolean constant or function call. Constants: true or false. Function call example: op_gt(id, 10). func_invoke: function call. Supported: e_drop, e_keep, e_set, e_if, e_compose	e_if(op_gt(id, 10), e_drop()); drops the record if ID > 10.
  if-else statement	e_if_else(bool_expr, func_invoke1, func_invoke2)	bool_expr: boolean constant or function call. Constants: true or false. Function call example: op_gt(id, 10). func_invoke1: function call executed if condition is true. func_invoke2: function call executed if condition is false.	e_if_else(op_gt(id, 10), e_set(tag, 'large'), e_set(tag, 'small')); sets tag to "large" if ID > 10, otherwise to "small".
  Switch-like statement that evaluates multiple conditions and executes the first matching operation. Executes a default operation if no conditions match.	s_switch(condition1, func1, condition2, func2, ..., default=default_func)	condition1: boolean constant or function call. Constants: true or false. Function call example: op_gt(id, 10). func_invoke: function call. Checks condition1; if true, executes this function and exits the switch. If false, checks the next condition. default_func: function call executed if all conditions are false.	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')).
  Combine multiple operations	e_compose(func1, func2, func3, ...)	func1: function call. Can be e_set, e_drop, e_if. func2: function call. Can be e_set, e_drop, e_if.	e_compose(e_set(str_col, 'test'), e_set(dt_col, dt_now())); sets str_col to "test" and dt_col to the current time.

• Data manipulation functions

  Function	Syntax	Parameters	Example
  Drop this record (do not synchronize)	e_drop()	None	e_if(op_gt(id, 10), e_drop()); drops records where ID > 10.
  Keep this record (synchronize to destination)	e_keep(condition)	condition: boolean expression	e_keep(op_gt(id, 1)); synchronizes only records where ID > 1.
  Set column value	e_set(col, val, NEW)	col: column name val: constant or function call. Must match col's data type. NEW: converts col to val's data type (optional) Importante If you omit NEW, do not include the preceding comma. Ensure data type compatibility to avoid task errors.	e_set(dt_col, dt_now()); sets dt_col to current time. e_set(col1, col2 + 1); sets col1 to col2 + 1. e_set(col1, 1, NEW); converts col1 to numeric type and sets it to 1.
  MongoDB field retention, field dropping, and field name mapping	e_expand_bson_value('*', 'fieldA',{"fieldB":"fieldC"})	: field names to retain; means all fields. fieldA: field names to drop. {"fieldB":"fieldC"}: field name mapping; fieldB is the source field name, fieldC is the destination field name. Nota Field name mapping is optional.	e_expand_bson_value("*", "_id,name"); writes all fields except _id and name to the destination.