DTS migration, synchronization, and subscription objects - Data Transmission Service: Central de Documentação do Alibaba Cloud

Utilize as operações de API do DTS para configure ou consultar objetos de migração, sincronização e assinatura. Este tópico abrange definições de objetos, parâmetros e exemplos de configuração.

Operações de API e parâmetros relacionados

API	Descrição
ConfigureDtsJob Configure a DTS subscription task	configure os objetos no parâmetro de solicitação `Dblist`.
Query the details of a DTS task Query the list of DTS tasks and the execution details of each task	Consulte os objetos no parâmetro de resposta `DbObject`.

Definição de objetos de migração, sincronização ou assinatura

Os parâmetros relacionados a objetos usam strings JSON. A seção a seguir descreve os parâmetros disponíveis.

Quando os objetos incluem vários bancos de dados:

Importante

As instâncias de assinatura não suportam mapeamento. O valor do parâmetro name deve corresponder ao nome real do banco de dados ou da tabela.

{
    "Name of database 1 to be migrated, synchronized, or subscribed to": {
        "name": "Name of database 1 in the destination instance",
        "all": true (Indicates that the entire database is migrated, synchronized, or subscribed to)
    },
    "Name of database 2 to be migrated, synchronized, or subscribed to": {
        "name": "Name of database 2 in the destination instance",
        "all": false (Indicates that the entire database is not migrated, synchronized, or subscribed to),
        "Table": {
            "Name of table A to be migrated, synchronized, or subscribed to": {
                "name": "Name of table A in the destination instance",
                "all": true (Indicates that the entire table is migrated, synchronized, or subscribed to),
                "dml_op": "DML operations to be incrementally migrated or synchronized",
                "ddl_op": "DDL operations to be incrementally migrated or synchronized"
            }
        }
    },
    "Name of database 3 to be migrated, synchronized, or subscribed to": {
        "name": "Name of database 3 in the destination instance",
        "all": true (Indicates that the entire database is migrated, synchronized, or subscribed to),
        "dml_op": "DML operations to be incrementally migrated or synchronized",
        "ddl_op": "DDL operations to be incrementally migrated or synchronized"
    }
}

Quando os objetos estão no nível de coluna ou incluem condições de filtro:

{
    "Name of database to be migrated, synchronized, or subscribed to": {
        "name": "Name of database in the destination instance",
        "all": false (Indicates that the entire database is not migrated, synchronized, or subscribed to),
        "Table": {
            "Name of table A to be migrated, synchronized, or subscribed to": {
                "name": "Name of table A in the destination instance",
                "all": false (Indicates that the entire table is not migrated, synchronized, or subscribed to),
                "filter": "id>10"
                "column": {
                    "id": {
                        "key": "PRI",
                        "name": "id",
                        "type": "int(11)",
                        "sharedKey": false,
                        "state": "checked"
                    }
                },
                "shard": 12
            }
        }
    }
}

Quando o destino é AnalyticDB for MySQL ou AnalyticDB for PostgreSQL:

{
    "Name of database to be migrated or synchronized": {
        "name": "Name of database in the destination instance",
        "all": false (Fixed as false. Regardless of whether the objects are at the database or table level, if the destination instance is AnalyticDB for MySQL or AnalyticDB for PostgreSQL, this parameter is fixed as false, and you must also specify information such as the partition key of the table),
        "Table": {
            "Name of table A to be migrated or synchronized": {
                "all": true (Indicates that the entire table is migrated or synchronized),
                "name": "Name of table A in the destination instance",
                "primary_key": "id (Specifies the primary key)",
                "type": "dimension (Type of the table)",
            }
            "Name of table B to be migrated or synchronized": {
                "all": true (Indicates that the entire table is migrated or synchronized),
                "name": "Name of table B in the destination instance",
                "part_key": "id (Specifies the partition key)",
                "primary_key": "id (Specifies the primary key)",
                "type": "partition (Type of the table)",
                "tagColumnValue": "Value of the tag column"
            }
        }
    }
}

Para defina políticas independentes de resolução de conflitos para objetos de sincronização:

Nota

Este recurso é compatível apenas com instâncias de sincronização bidirecional entre instâncias MySQL ou entre clusters PolarDB for MySQL.
defina políticas independentes de resolução de conflitos no nível do banco de dados ou da tabela.
Para colunas configuradas com uma política independente de resolução de conflitos, a política global de resolução de conflitos não tem efeito.

Configurações no nível da tabela

{
    "Name of database 1 to be synchronized": {
      "name": "Name of database 1 in the destination instance",
      "all": true (Indicates that the entire database is synchronized),
      "conflict": "Task-level conflict resolution policy"
    },
    "Name of database 2 to be synchronized": {
      "name": "Name of database 2 in the destination instance",
      "all": false (Indicates that the entire database is not synchronized),
      "conflict": "overwrite",
      "Table": {
        "Name of table A to be synchronized": {
          "name": "Name of table A in the destination instance",
          "all": true (Indicates that the entire table is synchronized),
          "cdr_cmp_col": "Conflict detection column",
          "cdr_rslv_col": "Conflict detection column",
          "resolve_method": "Table-level conflict resolution policy"
        }
      }
    }
}

Configurações no nível do banco de dados

Quando o objeto de sincronização é um banco de dados inteiro:

"Name of database 1 to be synchronized": {
  "name": "Name of database 1 in the destination instance",
  "all": true (Indicates that the entire database is synchronized),
  "conflict": "Task-level conflict resolution policy",
  "cdr_cmp_col": "Conflict detection column",
  "cdr_rslv_col": "Conflict detection column",
  "resolve_method": "Database-level conflict resolution policy"
  }
}

Quando o objeto de sincronização não é um banco de dados inteiro:

"Name of database 2 to be synchronized": {
  "name": "Name of database 2 in the destination instance",
  "all": false (Indicates that the entire database is not synchronized),
  "conflict": "Task-level conflict resolution policy",
  "cdr_cmp_col": "Conflict detection column",
  "cdr_rslv_col": "Conflict detection column",
  "resolve_method": "Database-level conflict resolution policy",
  "Table": {
    "Name of table A to be synchronized": {
      "name": "Name of table A in the destination instance",
      "all": true (Indicates that the entire table is synchronized)
    }
  }
}

Ao configurar uma tarefa de integração de dados para um data lake:

Parâmetro	Descrição
`write_operation`	O método de gravação quando ocorre um conflito de dados. `append`: Mantém os dados existentes e adiciona novos dados. `overwrite`: Sobrescreve os dados conflitantes no destino. `errorIfExists`: A tarefa gera um erro e é encerrada. `ignore`: Ignora a gravação atual e mantém os dados existentes no destino.
`targetType`	O formato dos dados após a gravação no OSS (conversão forçada). Formatos suportados: `Byte`, `Integer`, `Long`, `Double`, `String`, `Binary`, `Boolean`, `Timestamp` e `Date`. Nota Se não especifique, o DTS converte automaticamente os tipos de dados source.
`etl_date`	O nome da coluna adicional (constante) a ser adicionada. Nota Os valores dos dois parâmetros `etl_date` devem ser iguais.
`syntacticType`	defina como ADD, que indica a adição de uma coluna. Nota O valor da coluna adicionada (`value`) deve ser uma constante entre aspas simples ('').
`part_key`	A chave de partição da tabela de destino. Dois valores possíveis: Nota Obrigatório apenas quando a tabela de destino é particionada. Uma coluna a ser integrada da source. Uma coluna constante adicionada ao destino e defina como chave de partição. Nota O formato é `<Key>=<Value>`. Por exemplo, `dt=2025-07-07` indica uma coluna constante chamada `dt` com o valor `2025-07-07`.

{
    "Name of database 1 to be integrated": {
        "all": false (Fixed as false), 
        "Table": {
            "Name of table A to be integrated": {
                "all": false (Indicates that the entire table is not integrated), 
                "filter": "", 
                "write_operation": "Method used to write data", 
                "name": "Name of table A in the destination instance", 
                "column": {
                    "Name of column a to be integrated": {
                        "name": "Name of column a in the destination instance", 
                        "targetType": "Type of the column in OSS"
                    }, 
                    ******, 
                    "etl_date": {
                        "syntacticType": "ADD (Fixed as ADD)", 
                        "name": "etl_date", 
                        "type": "String (Fixed as String)", 
                        "value": "'2025-07-08 03:30:00'"
                    }
                }, 
                "part_key": "dt=2025-07-07"
            }
        }, 
        "name": "dtstestdata (Name of database 1 in the destination instance)"
    }
}

Parâmetro	Descrição
`name`	O nome mapeado do banco de dados, tabela ou coluna source no destino. Por exemplo, para migrar o banco de dados dtssource para dtstarget, defina este parâmetro como dtstarget.
`all`	Indica se você selecione todas as tabelas ou colunas. Valores válidos: `true`: sim. Nota Se definido como true, não é necessário especifique tabelas ou colunas individuais. `false`: não.
`Table`	As informações da tabela source.
`filter`	A condição de filtro para os dados a serem migrados, sincronizados ou assinados. Só pode ser defina no nível da tabela. Por exemplo, defina este parâmetro como `id>10` para migrar ou sincronizar apenas os dados cujo valor da coluna ID seja maior que 10. Para obter mais informações sobre o formato das condições de filtro, consulte Filter task data with an SQL condition. Nota As tarefas de assinatura não suportam condições de filtro.
`column`	As informações da coluna source.
`key`	Especifica se a coluna é uma chave primária. Valores válidos: `PRI`: sim. String vazia: não.
`sharedKey`	Especifica se a coluna é uma chave de shard. Valores válidos: `true`: sim. `false`: não. Nota Obrigatório apenas quando o tipo de banco de dados é Kafka.
`type`	O tipo de dados do campo.
`state`	Se o valor for `checked`, selecione a coluna.
`shard`	A contagem de shards para a tabela a ser migrada ou sincronizada. Nota Obrigatório apenas quando o tipo de banco de dados é Kafka.
`dml_op`	Operações DML para migrar ou sincronizar incrementalmente. Valores válidos: `i`: INSERT. `u`: atualize. `d`: exclua. Se este parâmetro estiver vazio, todas as operações DML suportadas pela tarefa são migradas ou sincronizadas incrementalmente. `none`: Nenhuma operação DML é migrada ou sincronizada incrementalmente. Nota Para consultar as operações DML suportadas por tipo de tarefa, consulte a documentação de configuração em Migration solutions ou Synchronization solutions.
`ddl_op`	Operações DDL para migrar ou sincronizar incrementalmente. Valores válidos: `ct`: crie TABLE. `at`: ALTER TABLE. `dt`: DROP TABLE. `rt`: RENAME TABLE. `tt`: TRUNCATE TABLE. Se este parâmetro estiver vazio, todas as operações DDL suportadas pela tarefa são migradas ou sincronizadas incrementalmente. `none`: Nenhuma operação DDL é migrada ou sincronizada incrementalmente. Nota Para consultar as operações DDL suportadas por tipo de tarefa, consulte a documentação de configuração em Migration solutions ou Synchronization solutions.
`primary_key`	A chave primária. Obrigatório apenas quando o destino é AnalyticDB for MySQL ou AnalyticDB for PostgreSQL.
`part_key`	A chave de partição. Obrigatório quando o destino é AnalyticDB for MySQL ou AnalyticDB for PostgreSQL.
`type` Importante Este parâmetro `type` é diferente do parâmetro `type` que representa o tipo de dados de um campo.	O tipo de tabela quando o destino é AnalyticDB for MySQL ou AnalyticDB for PostgreSQL. Valores válidos: dimension: tabela de dimensões. partition: tabela particionada.
`tagColumnValue`	O valor personalizado da coluna de tag __dts_data_source. Obrigatório quando o destino é AnalyticDB for MySQL.
`conflict`	A política global de resolução de conflitos no nível da tarefa. Ela deve ser incluída com o mesmo valor em cada banco de dados a ser sincronizado. Valores válidos: `overwrite`: Sobrescreve os registros conflitantes no banco de dados de destino. `interrupt`: Gera um erro e é encerrada. A tarefa entra no estado de falha e exige intervenção manual. Nota O console exibe TaskFailed. `ignore`: Ignora a instrução conflitante e mantém os registros existentes no banco de dados de destino.
`resolve_method`	A política independente de resolução de conflitos no nível da tabela. Compatível apenas com sincronização incremental. Valores válidos: `overwrite`: Sobrescreve os registros conflitantes no banco de dados de destino. `interrupt`: Gera um erro e é encerrada. A tarefa entra no estado de falha e exige intervenção manual. Nota O console exibe TaskFailed. `ignore`: Ignora a instrução conflitante e mantém os registros existentes no banco de dados de destino. `use_max`: Compara os registros conflitantes e grava aquele com o maior valor. Adota `overwrite` como fallback se o registro de destino não existir ou se o tipo de campo não for suportado. `use_min`: Compara os registros conflitantes e grava aquele com o menor valor. Adota `ignore` como fallback se o registro de destino não existir ou se o tipo de campo não for suportado.
`cdr_cmp_col`	Nível da tabela: A coluna de detecção de conflitos para a qual se deve defina uma política independente de resolução de conflitos, excluindo chaves primárias e únicas. Os valores devem ser iguais. Importante Quando a política de resolução de conflitos é `use_max` ou `use_min`, é obrigatório especifique os parâmetros `cdr_cmp_col` e `cdr_rslv_col`. O valor já inclui chaves primárias e chaves únicas por padrão. Não especifique manualmente as colunas correspondentes. Nível do banco de dados: A coluna de detecção de conflitos para a qual se deve defina uma política independente de resolução de conflitos. Os valores devem ser iguais. Importante É obrigatório especifique os parâmetros `cdr_cmp_col` e `cdr_rslv_col`.
`cdr_rslv_col`

Exemplos de configuração

Exemplo 1: Migrar, sincronizar ou assinar todas as tabelas no banco de dados dtstestdata.
```
{"dtstestdata": {   "name": "dtstestdata",   "all": true }}
```
Exemplo 2: Migrar ou sincronizar o banco de dados dtstestdata e renomeá-lo para dtstestdata_new.
```
{"dtstestdata": {   "name": "dtstestdata_new",   "all": true }}
```

Exemplo 3: Migrar, sincronizar ou assinar tabelas específicas (como customer) no banco de dados dtstestdata.

{"dtstestdata": {
   "name": "dtstestdata",
   "all": false,
   "Table": {
     "customer": {
       "name": "customer",
       "all": true, 
       "column": { 
         "id": {
           "key": "PRI",
           "name": "id",
           "type": "int(11)",
           "sharedKey": false,
           "state": "checked"  
         },
         "gmt_create": {
           "key": "",
           "name": "gmt_create",
           "type": "datetime",
           "sharedKey": false,
           "state": "checked"
         },
         "gmt_modify": {
           "key": "",
           "name": "gmt_modify",
           "type": "datetime",
           "sharedKey": false,
           "state": "checked"
         },
         "valid_time": {
           "key": "",
           "name": "valid_time",
           "type": "datetime",
           "sharedKey": false,
           "state": "checked"
         },
         "creator": {
           "key": "",
           "name": "creator",
           "type": "varchar(200)",
           "sharedKey": false,
           "state": "checked"
         }
       },
       "shard": 12
     }
   }
 }
}

Exemplo 4: Migrar ou sincronizar colunas específicas de tabelas (como customer e order) no banco de dados dtstestdata.

{"dtstestdata": {
   "name": "dtstestdata",
   "all": false,
   "Table": {
     "customer": {
       "name": "customer",
       "all": false, 
       "column": { 
         "id": {
           "key": "PRI",
           "name": "id",
           "type": "int(11)",
           "sharedKey": false,
           "state": "checked"  
         },
         "level": {
           "key": "",
           "name": "level",
           "type": "varchar(5000)",
           "sharedKey": false,
           "state": "checked"
         },
         "name": {
           "key": "",
           "name": "name",
           "type": "varchar(500)",
           "sharedKey": false,
           "state": "checked"
         },
       },
       "shard": 12
     },
     "order": {
       "name": "order",
       "all": false,
      "column": {
         "id": {
           "key": "PRI",
           "name": "id",
           "type": "int(11)",
           "sharedKey": false,
           "state": "checked"
         }
       },
       "shard": 12
     }
   }
 }
}

Exemplo 5: Migrar ou sincronizar tabelas (como customer, order e commodity) do banco de dados dtstestdata para uma instância de destino AnalyticDB for MySQL ou AnalyticDB for PostgreSQL.

{
    "dtstestdata": {
        "name": "dtstestdatanew",
        "all": false,
        "Table": {
            "order": {
                "name": "ordernew",
                "all": true,
                "part_key": "id",
                "primary_key": "id",
                "type": "partition"
            },
            "customer": {
                "name": "customernew",
                "all": true,
                "primary_key": "id",
                "type": "dimension"
            },
            "commodity": {
                "name": "commoditynew",
                "all": false,
                "filter": "id>10",
                "column": {
                    "id": {
                        "key": "PRI",
                        "name": "id",
                        "type": "int(11)"
                    }
                },
                "part_key": "id",
                "primary_key": "id",
                "type": "partition"
            }
        }
    }
}

Exemplo 6: defina políticas independentes de resolução de conflitos para objetos de sincronização.
Configurações no nível da tabela

defina a política global de resolução de conflitos para os objetos da tarefa de sincronização como interrupt. defina a política independente de resolução de conflitos para as colunas de chave primária, colunas de chave única e a coluna name da tabela customer no banco de dados dtstestdata2 como overwrite.
```
{
    "dtstestdata1": {
      "name": "dtstestdata1",
      "all": true,
      "conflict": "interrupt"
    },
    "dtstestdata2": {
      "name": "dtstestdata2",
      "all": false,
      "conflict": "interrupt",
      "Table": {
        "customer": {
          "name": "customer",
          "all": true,
          "cdr_cmp_col": "name",
          "cdr_rslv_col": "name",
          "resolve_method": "overwrite"
        }
      }
    }
  }
```
Configurações no nível do banco de dados
- Quando o objeto de sincronização é um banco de dados inteiro: defina a política independente de resolução de conflitos para as colunas name e addr de todas as tabelas a serem sincronizadas no banco de dados dtstestdata1 como use_max.
```
"dtstestdata1": {
  "name": "dtstestdata1",
  "all": true,
  "conflict": "overwrite",
  "cdr_cmp_col": "name,addr",
  "cdr_rslv_col": "name,addr",
  "resolve_method": "use_max"
  }
}
```
- Quando o objeto de sincronização não é um banco de dados inteiro: defina a política independente de resolução de conflitos para as colunas name e addr de todas as tabelas a serem sincronizadas no banco de dados dtstestdata2 como use_max.
```
"dtstestdata2": {
  "name": "dtstestdata2",
  "all": false,
  "conflict": "overwrite",
  "cdr_cmp_col": "name,addr",
  "cdr_rslv_col": "name,addr",
  "resolve_method": "use_max",
  "Table": {
    "person": {
      "name": "person",
      "all": true
    },
    "class": {
      "name": "class",
      "all": true
    }
  }
}
```

Exemplo 7: Para uma instância de sincronização cujo tipo de banco de dados source é Tair/Redis, sincronize apenas dados com o prefixo de chave HProp dos DBs 0 e 1 (ou seja, o campo Prefixes of Keys to Be Synchronized é HProp). Para o DB 2, sincronize apenas dados com o prefixo de chave dts, mas sem incluir dtstest (ou seja, o campo Prefixes of Keys to Be Synchronized é dts e o campo Prefixes of Keys to Be Filtered Out é dtstest).

{
    "0": {
        "name": "0", 
        "all": true,
         "filter": "[{"condition":"HProp","filterType":"white","filterPattern":"prefix"}]"
    }, 
    "1": {
        "name": "1", 
        "all": true,
         "filter": "[{"condition":"HProp","filterType":"white","filterPattern":"prefix"}]"
    }, 
    "2": {
        "name": "2", 
        "all": true,
         "filter": "[{"condition":"dts","filterType":"white","filterPattern":"prefix"},{"condition":"dtstest","filterType":"black","filterPattern":"prefix"}]"
    }
}

Exemplo 8: Integrar a tabela commodity do banco de dados dtstestdata ao OSS de destino no formato Delta.

{
    "dtstestdata(": {
        "all": false, 
        "Table": {
            "commodity": {
                "all": false(), 
                "filter": "", 
                "write_operation": "overwrite", 
                "name": "commodity", 
                "column": {
                    "IS_VALID": {
                        "name": "is_valid", 
                        "targetType": "String"
                    }, 
                    "BuiltinArchiveDate": {
                        "name": "builtinarchivedate", 
                        "targetType": "String"
                    }, 
                    "PRODUCT_NAME": {
                        "name": "product_name", 
                        "targetType": "String"
                    }, 
                    "PRODUCT_CODE": {
                        "name": "product_code", 
                        "targetType": "String"
                    }, 
                    "etl_date": {
                        "syntacticType": "ADD", 
                        "name": "etl_date", 
                        "type": "String", 
                        "value": "'2025-07-08 03:30:00'"
                    }
                }, 
                "part_key": "dt=2025-07-07"
            }
        }, 
        "name": "dtstestdata"
    }
}