文档中心

大数据研发治理套件

Activity （任务节点）类型

Copy Data

本文介绍 Copy Data 类型的 Activity 配置，用于在 Pipeline 中执行数据集成任务，将数据从源端同步到目标端。

概述

Copy Data Activity（YAML 类型标识为 dts_batch）在 Pipeline 中执行数据搬运任务，底层基于 DataSail 引擎实现异构数据源间的高效数据传输。
适用场景：

将 SQLServer、PostgreSQL 等业务库数据同步到数据仓库（Hive）。
跨数据库的表级数据迁移。
多表批量同步（动态映射模式）。

支持的数据源类型

Reader（数据源）

type 值	数据源	说明
`hive`	Hive	离线数仓表
`pg`	PostgreSQL	关系型数据库
`sql_server`	SQL Server	关系型数据库
`tos`	TOS	对象存储文件
`iceberg`	Iceberg	开放表格式数据湖表

Writer（目标端）

type 值	目标端	说明
`hive`	Hive	离线数仓表
`pg`	PostgreSQL	关系型数据库
`sql_server`	SQL Server	关系型数据库
`tos`	TOS	对象存储文件
`iceberg`	Iceberg	开放表格式数据湖表

创建 Copy Data 文件

单击新建 > CopyData 脚本按钮，创建 CopyData 脚本。
创建 CopyData 脚本后，进入 CopyData 配置页面。
在左上角切换“页面配置模式”或“文件配置模式”。

页面配置模式

选择数据源与数据目标

选择需要同步的数据源与目标类型。
选择对应数据源、数据目标，数据源设置详见配置数据源。

数据映射

Copy Data 支持以下两种表映射模式：

选表映射：手动选择要同步的数据表，并指定字段映射规则，同步数据，适用于相对固定的同步场景。
动态映射：支持参数或变量指定数据源表，动态匹配或新建目标表，适用于自动化 Pipeline 流程。

说明

当 Hive 作为来源端时，暂不支持 ZSTD 压缩的 ORC 文件。
TOS 的表映射信息与其他数据源类型不同，相关参数说明请参见配置 TOS 任务。

选表映射

您可通过以下几种方式操作字段映射关系：

单击自动添加，可以根据数据源 Schema 自动添加字段。部分数据源类型支持获取schema自动添加字段。
单击手动添加，可以添加一条空白的字段信息，手工输入字段名、类型等。
单击同名映射，可根据目标端/源端的字段添加情况，进行同名映射添加源端/目标端的字段信息。
说明
仅部分数据源支持同名映射能力，如 FTP/SFTP 数据源。
单击删除全部，可以删除全部字段信息。
通过拖拽字段左侧的按钮，可以调整字段顺序。

注意

请注意列与列之间映射的字段类型是否兼容。
尽量保持数据源表字段与目标表字段在映射中可以一一对应。

动态映射

表映射配置

源表：可以以常量或变量的形式输入数据源表名称，支持的参数变量配置详见参数配置规则。
匹配目标表：会按照下方转换规则配置对数据表做匹配，举例：
```
数据源表：user_info
目标表：ods_user_info_i_d
在目标表名的配置中填写：ods_${src_table_name}_i_d
则可以完成匹配
```
当无法匹配时，会按照配置新建目标表，或者报错。

字段映射配置

映射策略支持按字段名称匹配或者指定映射。
选择手动指定映射时，支持输入变量/参数或常量，样例：

[
  { "sourceName": "id", "targetName": "id" },
  { "sourceName": "user_name", "targetName": "name" },
  { "sourceName": "create_time", "targetName": "created_at" }
]

读取与写入设置

对于不同的数据源和目标，其读取和写入设置不同（如下图所示）。以 Hive→PostgreSQL 为例，具体如下：

Hive 数据源的读取设置：设置读取的分区，支持数据参数或变量，如系统变量 ${DATE} 指定读取的分区，或其他参数与变量。
PostgreSQL 的写入设置：
- 写入前、写入后准备语句：支持输入 SQL，执行写入前清表等操作。
- 数据写入方式：指定写入行为 insert into、insert on duplicate。

执行设置

并发设置：由于资源原因或者任务本身特性等原因，实际执行时并发数可能小于等于此值。
脏数据设置：表示脏数据的最大容忍条数/比例。如果配置0，表示不允许脏数据存在。
自定义参数设置：指定数据同步过程中的更多行为，详见：高级参数配置总览。
资源组设置：选择执行任务的独享集成资源组，资源组开通相关配置详见：数据集成资源组。

文件配置模式

Copy Data 的详细配置以 JSON 文件形式存储在工作区中，文件命名规范为 <任务名>.copydata.json。

整体结构

数据同步任务的 JSON 顶层通常只有一个核心对象：

{
  "configuration": {
    "reader": {},
    "writer": {},
    "operator": {}
  }
}

其中：

一级区块	作用	典型内容
configuration.reader	描述从哪里读数据	源端类型、源数据源、源表、分区、源字段列表
configuration.writer	描述写到哪里	目标端类型、目标数据源、目标库/schema、目标表、写入模式、目标字段列表
configuration.operator	描述任务运行和映射规则	并发、资源组、自定义参数、动态 mapping 参数

reader 结构

reader 固定包含两层：

{
  "reader": {
    "type": "hive",
    "parameter": {}
  }
}

字段	说明
`reader.type`	源端连接器类型，例如 `hive`、`pg`、`sql_server`。
`reader.parameter`	源端连接器参数，不同数据源类型字段会有差异。

常见 reader.parameter 字段：

字段	说明
`dataSourceId`	源端数据源 ID。
`dataSourceName`	源端数据源名称。
`tableName`	源表名。
`readerMode`	读取模式，例如 Hive 的 `hdfs`。
`multiplePartitionList`	分区读取条件。
`columns`	源端字段列表。手动配置模式下通常非空；动态 Mapping 模式下可为空。
`connectorType`	连接器类型标识。

writer 结构

writer 同样固定包含两层：

{
  "writer": {
    "type": "pg",
    "parameter": {}
  }
}

字段	说明
`writer.type`	目标端连接器类型，例如 `pg`、`sql_server`、`hive`。
`writer.parameter`	目标端连接器参数，不同目标类型字段会有差异。

常见 writer.parameter 字段：

字段	说明
`dataSourceId`	目标端数据源 ID。
`dataSourceName`	目标端数据源名称。
`tableSchema` / `schemaName`	目标 schema。PostgreSQL 常用 `tableSchema`，SQL Server 常用 `schemaName`。
`tableName`	目标表名。动态 Mapping 模式下可以为空，由规则生成。
`writeMode`	写入模式，例如 `directlyInsert`、`overwrite`。
`preSqls`	写入前执行的 SQL。
`postSqls`	写入后执行的 SQL。
`columns`	目标端字段列表。手动配置模式下通常非空；动态 Mapping 模式下可为空。
`connectorType`	连接器类型标识。

operator 结构

operator 描述任务级配置，主要包括运行参数、资源组和映射参数：

{
  "operator": {
    "parameter": {
      "expectedMaxParallelism": 1,
      "customParameters": {},
      "extendedParameters": {},
      "resourceGroup": {}
    }
  }
}

字段	说明
`expectedMaxParallelism`	期望最大并发。
`customParameters`	自定义参数区。
`extendedParameters`	扩展参数区，动态 Mapping 参数通常放在这里。
`resourceGroup`	任务运行资源组。

动态 Mapping 相关参数可以放在以下任一区域：

extendedParameters
customParameters
internalParameters

实际读取优先级为：

extendedParameters > customParameters > internalParameters

因此对直接配置文件而言，推荐统一放在 operator.parameter.extendedParameters 中。

字段列表结构

reader.parameter.columns 和 writer.parameter.columns 的每个字段对象通常包含：

{
  "type": "bigint",
  "name": "lead_id",
  "extraType": null,
  "description": null
}

字段	说明
`name`	字段名。
`type`	字段类型。
`extraType`	扩展类型信息，通常可为空。
`description`	字段描述，通常可为空。

在手动配置映射模式下，reader 和 writer 的 columns 都由用户明确填写，并按数组顺序一一对应。
在动态 Mapping 模式下，reader 和 writer 的 columns 可以为空，系统会根据表元数据和动态规则生成最终字段列表。

字段列表结构

reader.parameter.columns 和 writer.parameter.columns 的每个字段对象通常包含：

{
  "type": "bigint",
  "name": "lead_id",
  "extraType": null,
  "description": null
}

字段	说明
`name`	字段名。
`type`	字段类型。
`extraType`	扩展类型信息，通常可为空。
`description`	字段描述，通常可为空。

页面和 JSON 的对应关系

数据同步配置页面可以理解为两层：

页面区域	主要作用	对应 JSON 区块
数据源和目标配置页	选择源端类型、目标端类型、数据源、表、分区、写入模式、资源组等	`configuration.reader`、`configuration.writer`、`configuration.operator.parameter.resourceGroup`
字段映射配置页	配置源端字段、目标端字段，以及字段如何对齐	`reader.parameter.columns`、`writer.parameter.columns`、`operator.parameter.extendedParameters`

在选表映射模式下，字段映射页面会把字段列表直接写入：

reader.parameter.columns
writer.parameter.columns

在动态映射模式下，字段映射页面不需要手动写完整字段列表，而是把动态规则写入：

operator.parameter.extendedParameters

系统在生成实例配置时，再根据这些规则补齐目标表名和字段列表。

模式	mappingType	表名配置方式	字段配置方式	适用场景
手动配置映射	`table_mapping`	用户明确填写源表和目标表。	用户明确填写 reader/writer 字段列表。	表结构明确、字段映射固定、希望完全控制同步字段
动态 Mapping	`dynamic_mapping`	系统根据源表名和规则生成目标表名。	系统根据源表和目标表元数据自动生成字段。	多表批量同步、目标表名有规则、希望减少手工配置

注意

当前后端动态 Mapping 逻辑的触发，主要依赖 columnMappingMode、tableNameMappingRules、tableMappingStrategy、columnNameMappingRules 等动态字段，而不是单独依赖 mappingType。
mappingType 更适合作为页面或业务侧区分配置模式的标识。

模式一：选表映射

样例说明

以 Hive -> SQL Server 为例：

Hive lead_mart -> SQL Server dbo.lead_mart

源端配置：

{
  "type": "hive",
  "parameter": {
    "dataSourceId": 55028,
    "dataSourceName": "EMR_Serverless_hive_dmp_poc",
    "tableName": "lead_mart",
    "readerMode": "hdfs",
    "multiplePartitionList": [
      [
        {
          "name": "dt",
          "type": "string",
          "value": "${DATE}"
        }
      ]
    ],
    "columns": [
      {
        "type": "bigint",
        "name": "lead_id"
      },
      {
        "type": "varchar(17)",
        "name": "vin"
      },
      {
        "type": "varchar(50)",
        "name": "lead_type"
      },
      {
        "type": "varchar(20)",
        "name": "dealer_code"
      },
      {
        "type": "varchar(20)",
        "name": "priority"
      },
      {
        "type": "timestamp",
        "name": "created_ts"
      },
      {
        "type": "varchar(30)",
        "name": "status"
      }
    ],
    "connectorType": "hive"
  }
}

目标端配置：

{
  "type": "sql_server",
  "parameter": {
    "dataSourceId": 54933,
    "dataSourceName": "sqlserver_test",
    "schemaName": "dbo",
    "tableName": "lead_mart",
    "writeMode": "overwrite",
    "columns": [
      {
        "type": "bigint",
        "name": "lead_id"
      },
      {
        "type": "varchar",
        "name": "vin"
      },
      {
        "type": "varchar",
        "name": "lead_type"
      },
      {
        "type": "varchar",
        "name": "dealer_code"
      },
      {
        "type": "varchar",
        "name": "priority"
      },
      {
        "type": "datetime",
        "name": "created_ts"
      },
      {
        "type": "varchar",
        "name": "status"
      }
    ],
    "connectorType": "sql_server"
  }
}

模式标识：

{
  "extendedParameters": {
    "mappingType": "table_mapping"
  }
}

选表映射的执行逻辑

选表映射的核心特点是：字段列表已经在 JSON 中明确给出，系统不需要再根据元数据推导字段。
源端字段和目标端字段按数组顺序一一对应：

序号	源字段	源类型	目标字段	目标类型
1	`lead_id`	`bigint`	`lead_id`	`bigint`
2	`vin`	`varchar(17)`	`vin`	`varchar`
3	`lead_type`	`varchar(50)`	`lead_type`	`varchar`
4	`dealer_code`	`varchar(20)`	`dealer_code`	`varchar`
5	`priority`	`varchar(20)`	`priority`	`varchar`
6	`created_ts`	`timestamp`	`created_ts`	`datetime`
7	`status`	`varchar(30)`	`status`	`varchar`

这类配置适用于：

客户已经明确知道同步哪些字段。
目标表已经存在，且目标字段已经确定。
需要控制字段顺序、字段类型或字段数量。
源字段和目标字段可能有差异，但希望通过页面手动确认。

选表映射的注意事项

选表映射要求：

reader.parameter.columns 不能为空。
writer.parameter.columns 不能为空。
reader 字段数量和 writer 字段数量需要一致。
字段顺序就是实际同步映射顺序。

注意

如果源端字段和目标端字段数量不一致，或顺序配置错误，可能导致写入字段错位或配置校验失败。

模式二：动态 Mapping

样例说明

以 Hive -> PostgreSQL 为例：

Hive lead_fact -> PostgreSQL public.lead_fact_depatch_queue

源端只指定源表：

{
  "type": "hive",
  "parameter": {
    "dataSourceId": 55028,
    "dataSourceName": "EMR_Serverless_hive_dmp_poc",
    "tableName": "lead_fact",
    "readerMode": "hdfs",
    "multiplePartitionList": [
      [
        {
          "name": "dt",
          "type": "string",
          "value": "${DATE}"
        }
      ]
    ],
    "columns": [],
    "connectorType": "hive"
  }
}

目标端不直接填写目标表名：

{
  "type": "pg",
  "parameter": {
    "dataSourceId": 54934,
    "dataSourceName": "postgres_test",
    "tableSchema": "public",
    "tableName": "",
    "writeMode": "directlyInsert",
    "columns": [],
    "connectorType": "pg"
  }
}

动态规则放在 extendedParameters：

{
  "extendedParameters": {
    "tableNameMappingRules": [
      {
        "sourceName": "${src_table_name}",
        "targetName": "${src_table_name}_depatch_queue"
      }
    ],
    "tableMappingStrategy": "prefer_existing_else_create",
    "columnMappingMode": "name_match",
    "nameMatchRule": "ignore_case",
    "onColumnMismatch": "ignore",
    "tableName": "lead_fact",
    "mappingType": "dynamic_mapping"
  }
}

动态 Mapping 的执行逻辑

动态 Mapping 的核心流程如下：

系统读取源表名 lead_fact。
系统发现目标表名为空。
系统读取 tableNameMappingRules。
${src_table_name} 被替换为 lead_fact。
目标表名生成结果为 lead_fact_depatch_queue。
系统检查 PostgreSQL 中 public.lead_fact_depatch_queue 是否存在。
如果目标表不存在，并且 tableMappingStrategy=prefer_existing_else_create，系统按源表结构创建目标表。
如果目标表已存在，系统根据字段映射规则生成 reader/writer 字段。

最终写入目标表：

public.lead_fact_depatch_queue

动态表名规则

tableNameMappingRules 用于将源表名转换成目标表名。

示例：

[
  {
    "sourceName": "${src_table_name}",
    "targetName": "${src_table_name}_depatch_queue"
  }
]

支持的 sourceName：
sourceName
含义
${src_table_name}
匹配当前源表名
*
匹配任意源表
match_all
匹配任意源表
具体表名
匹配指定源表名，忽略大小写。
支持的 targetName 变量：
变量
含义
${src_table_name}
源表名
${src_db_name}
源库名
${src_schema_name}
源 schema 名

sourceName	含义
`${src_table_name}`	匹配当前源表名
`*`	匹配任意源表
`match_all`	匹配任意源表
具体表名	匹配指定源表名，忽略大小写。

变量	含义
`${src_table_name}`	源表名
`${src_db_name}`	源库名
`${src_schema_name}`	源 schema 名

如果 writer.parameter.tableName 已经有值，系统会优先使用该值，不会再用 tableNameMappingRules 覆盖。

目标表不存在时的策略

tableMappingStrategy 控制目标表不存在时的处理方式。

值	行为
`prefer_existing_else_create`	优先使用已有目标表；如果不存在，则自动创建目标表。
`prefer_existing_else_error`	优先使用已有目标表；如果不存在，则报错。

建议：

初始化新目标表时使用 prefer_existing_else_create。
生产环境目标表由 DBA 或数仓团队管控时使用 prefer_existing_else_error。

字段映射模式

columnMappingMode 控制字段如何映射。

值	行为
`name_match`	按字段名自动匹配
`manual`	按 `columnNameMappingRules` 手动指定字段映射

如果目标表是本次自动创建的，系统会直接基于源表字段生成目标字段，不再额外做字段名匹配。
如果目标表已经存在，系统会根据 columnMappingMode 进行字段映射。

按字段名自动匹配

配置示例：

{
  "columnMappingMode": "name_match",
  "nameMatchRule": "ignore_case",
  "onColumnMismatch": "ignore"
}

含义：

源字段和目标字段按字段名匹配。
字段名大小写可以忽略。
匹配不到的源字段跳过。

示例：

源字段	目标字段	结果
`ID`	`id`	匹配
`name`	`NAME`	匹配
`age`	无	跳过

字段名匹配规则（严格匹配或忽略大小写）

值	行为
`ignore_case`	忽略大小写，`ID` 和 `id` 视为同一字段。
`strict`	严格匹配，字段名必须完全一致。

示例：

源字段	目标字段	`ignore_case`	`strict`
`ID`	`id`	匹配	不匹配
`name`	`name`	匹配	匹配

目标字段匹配不到时处理策略

值	行为
`ignore`	跳过不匹配字段，只同步匹配成功的字段。
`error`	发现不匹配字段时报错。

建议：

如果客户接受部分字段同步，使用 ignore。
如果客户要求源表和目标表字段严格一致，使用 error。

动态映射模式下的手动字段映射

动态 Mapping 也支持手动指定字段映射。这个模式和“手动配置映射”不同：

手动配置映射：直接填写完整的 reader.columns 和 writer.columns。
动态 Mapping 的 manual：不直接填写完整字段列表，而是填写字段映射规则，系统仍会查询源表和目标表元数据后生成最终字段列表。

配置示例：

{
  "columnMappingMode": "manual",
  "nameMatchRule": "strict",
  "onColumnMismatch": "error",
  "columnNameMappingRules": [
    {
      "sourceName": "id",
      "targetName": "user_id"
    },
    {
      "sourceName": "name",
      "targetName": "user_name"
    }
  ]
}

含义：

源字段	目标字段
`id`	`user_id`
`name`	`user_name`

如果 onColumnMismatch=error，只要规则里的源字段或目标字段不存在，就会报错。
如果 onColumnMismatch=ignore，不存在的映射关系会被跳过。

旧版字段对齐模式

系统还兼容旧版字段对齐模式。

值	行为
`source_only` / `source` / `by_source`	以源表字段为准生成字段列表
`target_only` / `target` / `by_target`	以目标表字段为准生成字段列表
`intersection` / `intersect` / `by_intersection`	只同步源表和目标表交集字段

建议新任务优先使用：

columnMappingMode=name_match
columnMappingMode=manual

旧版模式主要用于兼容已有任务。

两种模式的关键差异

对比项	手动配置映射	动态 Mapping
目标表名	用户直接填写 `writer.parameter.tableName`。	可通过 `tableNameMappingRules` 自动生成。
源字段	用户填写 `reader.parameter.columns`。	可留空，由系统读取源表元数据生成。
目标字段	用户填写 `writer.parameter.columns`。	可留空，由系统读取目标表元数据或自动建表后生成。
目标表不存在	通常要求用户提前准备目标表。	可配置自动建表或直接报错。
字段不一致	用户在页面上手动处理。	可配置忽略或报错。
适用表数量	更适合单表或少量表。	更适合规则化、多表批量场景。
可控性	高，用户完全指定字段。	高度自动化，依赖元数据和规则。

在 Pipeline 中添加 Copy Data 节点

在 Pipeline 中添加节点，选择 Copy Data 类型。
选择已配置好的 Copy Data 文件即可。

Pipeline YAML 配置

Copy Data 在 Pipeline YAML 中通过 taskName 引用预定义的集成任务配置：

- name: sync_orders
  type: dts_batch
  taskName: order_sync_task
  retryPolicy:
    maxRetries: 3
    minRetryIntervalMillis: 120000
  position:
    x: "200"
    y: "100"

字段说明

字段	类型	必填	说明
`type`	String	是	固定为 `dts_batch`。
`taskName`	String	是	数据集成任务名称，对应 `.copydata.json` 配置文件。

Pipeline 中的典型用法

For Each 与 Copy Data 连用，通过参数判断是否同步某张表的数据。

For Each 内部配置如下图所示：

使用建议

建议	说明
选择合适的映射模式	少量表（< 5）用手动映射精细控制；多表批量同步用动态映射提效。
动态映射必设 onColumnMismatch	建议设为 `ignore`，避免因源端新增字段导致任务失败。
合理设置并行度	`expectedMaxParallelism` 根据数据量和目标端承载能力调整。
开启脏数据清洗	生产环境建议开启 `dirtyRecordsCleansing` 并设置合理阈值。
配置文件命名规范	使用 `<业务描述>.copydata.json` 格式，便于识别和管理。
动态映射表名规则	善用 `${src_table_name}` 变量和正则，减少重复配置。
字段类型兼容性	注意源端和目标端的类型映射，避免精度丢失（如 DECIMAL → STRING）。
增量同步条件	Reader 的 `condition` 字段配合 `${date}` 变量实现每日增量。
配置重试策略	数据同步可能因网络波动失败，建议配置 retryPolicy。

最近更新时间：2026.08.26 17:59:08

这个页面对您有帮助吗？

有用

无用

大数据研发治理套件

概述 #

支持的数据源类型 #

Reader（数据源） #

Writer（目标端） #

创建 Copy Data 文件 #

页面配置模式 #

选择数据源与数据目标 #

数据映射 #

选表映射

动态映射

读取与写入设置 #

执行设置 #

文件配置模式 #

整体结构 #

reader 结构

writer 结构

operator 结构

字段列表结构

字段列表结构

页面和 JSON 的对应关系 #

模式一：选表映射 #

样例说明

选表映射的执行逻辑

选表映射的注意事项

模式二：动态 Mapping #

样例说明

动态 Mapping 的执行逻辑

动态表名规则

目标表不存在时的策略

字段映射模式

按字段名自动匹配

字段名匹配规则（严格匹配或忽略大小写）

目标字段匹配不到时处理策略

动态映射模式下的手动字段映射

旧版字段对齐模式

两种模式的关键差异 #

在 Pipeline 中添加 Copy Data 节点 #

Pipeline YAML 配置 #

Pipeline 中的典型用法 #

使用建议 #

概述

支持的数据源类型

Reader（数据源）

Writer（目标端）

创建 Copy Data 文件

页面配置模式

选择数据源与数据目标

数据映射

读取与写入设置

执行设置

文件配置模式

整体结构

页面和 JSON 的对应关系

模式一：选表映射

模式二：动态 Mapping

两种模式的关键差异

在 Pipeline 中添加 Copy Data 节点

Pipeline YAML 配置

Pipeline 中的典型用法

使用建议