DTS migration, synchronization, and subscription objects - Data Transmission Service

Llame a las operaciones de la API de DTS para configurar o consultar objetos de migración, sincronización y suscripción. Este tema abarca las definiciones de los objetos, los parámetros y los ejemplos de configuración.

Operaciones de la API y parámetros relacionados

API	Descripción
ConfigureDtsJob Configure a DTS subscription task	Configure los objetos en el parámetro de solicitud `Dblist`.
Query the details of a DTS task Query the list of DTS tasks and the execution details of each task	Consulte los objetos en el parámetro de respuesta `DbObject`.

Definición de objetos de migración, sincronización o suscripción

Los parámetros relacionados con los objetos utilizan cadenas JSON. A continuación se describen los parámetros disponibles.

Cuando los objetos incluyen varias bases de datos:

Importante

Las instancias de suscripción no admiten asignaciones. El valor del parámetro name debe coincidir con el nombre real de la base de datos o de la tabla.

{
    "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"
    }
}

Cuando los objetos están a nivel de columna o incluyen condiciones 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
            }
        }
    }
}

Cuando el destino es AnalyticDB for MySQL o 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"
            }
        }
    }
}

Cuando establece políticas de resolución de conflictos independientes para los objetos de sincronización:

Nota

Esta función solo es compatible con instancias de sincronización bidireccional entre instancias MySQL o entre clústeres de PolarDB for MySQL.
Es posible establecer políticas de resolución de conflictos independientes a nivel de base de datos o de tabla.
En las columnas configuradas con una política de resolución de conflictos independiente, la política global no surte efecto.

Configuración a nivel de tabla

{
    "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"
        }
      }
    }
}

Configuración a nivel de base de datos

Cuando el objeto de sincronización es una base de datos completa:

"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"
  }
}

Cuando el objeto de sincronización no es una base de datos completa:

"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)
    }
  }
}

Cuando configura una tarea de integración de datos hacia un lago de datos:

Parámetro	Descripción
`write_operation`	El método de escritura cuando ocurre un conflicto de datos. `append`: conserva los datos existentes y agrega los nuevos. `overwrite`: sobrescribe los datos en conflicto en el destino. `errorIfExists`: la tarea reporta un error y finaliza. `ignore`: omite la escritura actual y conserva los datos existentes en el destino.
`targetType`	El formato de los datos después de escribirlos en OSS (conversión obligatoria). Formatos compatibles: `Byte`, `Integer`, `Long`, `Double`, `String`, `Binary`, `Boolean`, `Timestamp` y `Date`. Nota Si no se especifica, DTS convierte automáticamente los tipos de datos de origen.
`etl_date`	El nombre de la columna adicional (constante) que se va a agregar. Nota Los valores de los dos parámetros `etl_date` deben ser iguales.
`syntacticType`	Fijo como ADD, lo cual indica que se agrega una columna. Nota El valor de la columna agregada (`value`) debe ser una constante encerrada entre comillas simples ('').
`part_key`	La clave de partición de la tabla de destino. Tiene dos valores posibles: Nota Obligatorio solo cuando la tabla de destino está particionada. Una columna que se integrará desde el origen. Una columna constante que se agrega al destino y se establece como clave de partición. Nota El formato es `<Key>=<Value>`. Por ejemplo, `dt=2025-07-07` indica una columna constante llamada `dt` con el 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	Descripción
`name`	El nombre asignado en el destino para la base de datos, la tabla o la columna de origen. Por ejemplo, para migrar la base de datos dtssource a dtstarget, establezca este parámetro como dtstarget.
`all`	Especifica si se seleccionan todas las tablas o columnas. Valores válidos: `true`: sí. Nota Si se establece en true, no es necesario especificar tablas ni columnas individuales. `false`: no.
`Table`	La información de la tabla de origen.
`filter`	La condición de filtro para los datos que se van a migrar, sincronizar o suscribir. Solo se puede establecer a nivel de tabla. Por ejemplo, puede establecer este parámetro en `id>10` para migrar o sincronizar únicamente los datos cuya columna ID tenga un valor mayor que 10. Para obtener más información sobre el formato de las condiciones de filtro, consulte Filter task data with an SQL condition. Nota Las tareas de suscripción no admiten la configuración de condiciones de filtro.
`column`	La información de la columna de origen.
`key`	Especifica si la columna es una clave primaria. Valores válidos: `PRI`: sí. Cadena vacía: no.
`sharedKey`	Especifica si la columna es una clave de fragmentación. Valores válidos: `true`: sí. `false`: no. Nota Obligatorio solo cuando el tipo de base de datos es Kafka.
`type`	El tipo de datos del campo.
`state`	Si el valor es `checked`, la columna está seleccionada.
`shard`	El número de fragmentos para la tabla que se va a migrar o sincronizar. Nota Obligatorio solo cuando el tipo de base de datos es Kafka.
`dml_op`	Las operaciones DML que se van a migrar o sincronizar de forma incremental. Valores válidos: `i`: INSERT. `u`: UPDATE. `d`: DELETE. Si este parámetro está vacío, se migran o sincronizan de forma incremental todas las operaciones DML que admite la tarea. `none`: no se migra ni sincroniza de forma incremental ninguna operación DML. Nota Para conocer las operaciones DML compatibles según el tipo de tarea, revise los documentos de configuración en Migration solutions o Synchronization solutions.
`ddl_op`	Las operaciones DDL que se van a migrar o sincronizar de forma incremental. Valores válidos: `ct`: CREATE TABLE. `at`: ALTER TABLE. `dt`: DROP TABLE. `rt`: RENAME TABLE. `tt`: TRUNCATE TABLE. Si este parámetro está vacío, se migran o sincronizan de forma incremental todas las operaciones DDL que admite la tarea. `none`: no se migra ni sincroniza de forma incremental ninguna operación DDL. Nota Para conocer las operaciones DDL compatibles según el tipo de tarea, revise los documentos de configuración en Migration solutions o Synchronization solutions.
`primary_key`	La clave primaria. Obligatorio solo cuando el destino es AnalyticDB for MySQL o AnalyticDB for PostgreSQL.
`part_key`	La clave de partición. Obligatorio cuando el destino es AnalyticDB for MySQL o AnalyticDB for PostgreSQL.
`type` Importante Este parámetro `type` es distinto del parámetro `type` que representa el tipo de datos de un campo.	El tipo de tabla cuando el destino es AnalyticDB for MySQL o AnalyticDB for PostgreSQL. Valores válidos: dimension: tabla de dimensiones. partition: tabla particionada.
`tagColumnValue`	El valor personalizado de la columna de etiqueta __dts_data_source. Obligatorio cuando el destino es AnalyticDB for MySQL.
`conflict`	La política global de resolución de conflictos a nivel de tarea. Debe incluirse en cada base de datos que se vaya a sincronizar con el mismo valor. Valores válidos: `overwrite`: sobrescribe los registros en conflicto en la base de datos de destino. `interrupt`: reporta un error y finaliza. La tarea entra en estado con errores y requiere intervención manual. Nota La consola muestra TaskFailed. `ignore`: omite la instrucción en conflicto y conserva los registros existentes en la base de datos de destino.
`resolve_method`	La política independiente de resolución de conflictos a nivel de tabla. Solo es compatible con la sincronización incremental. Valores válidos: `overwrite`: sobrescribe los registros en conflicto en la base de datos de destino. `interrupt`: reporta un error y finaliza. La tarea entra en estado con errores y requiere intervención manual. Nota La consola muestra TaskFailed. `ignore`: omite la instrucción en conflicto y conserva los registros existentes en la base de datos de destino. `use_max`: compara los registros en conflicto y escribe el que tiene el valor mayor. Recurre a `overwrite` si el registro de destino no existe o si el tipo de campo no es compatible. `use_min`: compara los registros en conflicto y escribe el que tiene el valor menor. Recurre a `ignore` si el registro de destino no existe o si el tipo de campo no es compatible.
`cdr_cmp_col`	A nivel de tabla: la columna de detección de conflictos para la cual se necesita establecer una política de resolución de conflictos independiente, sin incluir claves primarias ni claves únicas. Los valores deben ser iguales. Importante Cuando la política de resolución de conflictos es `use_max` o `use_min`, es obligatorio especificar los parámetros `cdr_cmp_col` y `cdr_rslv_col`. De manera predeterminada, el valor ya incluye las claves primarias y las claves únicas. No es necesario especificar manualmente las columnas correspondientes. A nivel de base de datos: la columna de detección de conflictos para la cual se necesita establecer una política de resolución de conflictos independiente. Los valores deben ser iguales. Importante Es obligatorio especificar los parámetros `cdr_cmp_col` y `cdr_rslv_col`.
`cdr_rslv_col`

Ejemplos de configuración

Ejemplo 1: migrar, sincronizar o suscribir todas las tablas de la base de datos dtstestdata.
```
{"dtstestdata": {   "name": "dtstestdata",   "all": true }}
```
Ejemplo 2: migrar o sincronizar la base de datos dtstestdata y cambiarle el nombre a dtstestdata_new.
```
{"dtstestdata": {   "name": "dtstestdata_new",   "all": true }}
```

Ejemplo 3: migrar, sincronizar o suscribir tablas específicas (como customer) de la base de datos 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
     }
   }
 }
}

Ejemplo 4: migrar o sincronizar columnas específicas de algunas tablas (como customer y order) de la base de datos 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
     }
   }
 }
}

Ejemplo 5: migrar o sincronizar tablas (como customer, order y commodity) de la base de datos dtstestdata hacia una instancia de destino AnalyticDB for MySQL o 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"
            }
        }
    }
}

Ejemplo 6: establecer políticas de resolución de conflictos independientes para los objetos de sincronización.
Configuración a nivel de tabla

Establezca la política global de resolución de conflictos para los objetos de la tarea de sincronización en interrupt. Luego, configure la política independiente de resolución de conflictos para las columnas de clave primaria, las columnas de clave única y la columna name de la tabla customer en la base de datos dtstestdata2 con el valor 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"
        }
      }
    }
  }
```
Configuración a nivel de base de datos
- Cuando el objeto de sincronización es una base de datos completa: establezca la política independiente de resolución de conflictos para las columnas name y addr de todas las tablas que se van a sincronizar en la base de datos dtstestdata1 con el valor use_max.
  "dtstestdata1": { "name": "dtstestdata1", "all": true, "conflict": "overwrite", "cdr_cmp_col": "name,addr", "cdr_rslv_col": "name,addr", "resolve_method": "use_max" } }
- Cuando el objeto de sincronización no es una base de datos completa: establezca la política independiente de resolución de conflictos para las columnas name y addr de todas las tablas que se van a sincronizar en la base de datos dtstestdata2 con el valor 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 } } }

Ejemplo 7: en una instancia de sincronización cuyo tipo de base de datos de origen es Tair/Redis, sincronice únicamente los datos con el prefijo de clave HProp de las bases de datos denominadas 0 y 1 (es decir, donde Prefixes of Keys to Be Synchronized es HProp). Para la base de datos denominada 2, sincronice solo los datos con el prefijo de clave dts, pero excluya dtstest (es decir, donde Prefixes of Keys to Be Synchronized es dts y Prefixes of Keys to Be Filtered Out es 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"}]"
    }
}

Ejemplo 8: integrar la tabla commodity de la base de datos dtstestdata en el destino OSS con 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"
    }
}