阿里云数据集成服务-深入介绍

1、数据集成进阶与深入数据集成/进阶与深入 PAGE 7进阶与深入为构建云上四通八达的数据同步通道，CDP Job自身定义一套数据中间格式。针对不同的数据通道，实现不同的读取插件完成从各类数据源抽取数据(CDP称之为Reader)，以及不同写入插件向各类数据源写入数据(CDP称 之为Writer)。CDP的各类插件实际上是完成了各类数据源对接CDP的适配器。如下：CDP利用RDSReader(实际上是MysqlReader)完成数据从RDS抽取，并转换为CDP的数据协议，投 递给CDP传输中间层。CDP利用ODPSWriter完成CDP的数据协议转换为ODPS的数据协议，并将转换后的数据向ODPS

2、写 入。通过上述CDP数据中间层，CDP可以为公有云上构建出四通八达的数据传输同步通道，让数据不再是孤岛！云数据库 OceanBase/相关协议CDP在进行数据同步过程中，为了提升Job数据传输吞吐能力，CDP通常将对传输数据集进行细粒度 切分(CDP称之为Task)，并启动多线程乃至于多进程运行容器运行Task进行数据传输服务。单个Task运行容器CDP定义为传输单元。CDP单个传输单元数据吞吐带宽为1MB/s，受限于源宿两端数据源的限制，CDP无法保证单个传输单 元一定维持在1MB/s。例如用户需要为单Job分配5MB/s的带宽，CDP将启动5个数据传输单元，吞吐 量维持在5MB/s左右。

3、由于两端数据源负载压力，该Job运行期间可能存在上下波动，无法一直维持 在5MB/s，但能够保证最大流量不超过5MB/s。为最大化数据同步的Job传输吞吐量，CDP将对Job进行更细粒度的任务切分，并启动多个数据传输单元进行并 行数据同步。CDP的切分规则由CDP框架定义接口，具体切分规则交由具体传输插件负责实现，例如：对于ODPS而言，数据切分根据ODPS表的Record偏移量计算数据分片。对于Mysql Reader插件而言，其切分算法按照Mysql数据表主键范围切分。以Mysql抽取数据为例，MysqlReader启动后将连接源数据端，计算数据表主键范围，如Min, Max，并根据用户带

4、宽需求切分上述Min, Max区间范围，每个子区间将作为一个Task提交给后续传输单元进行执行。例如，上述Mysql数据表主键范围在1,4，MysqlReader根据用户需要的吞吐量计算得出数据切分规则需要两个数 据子集，因此将数据区间区分为1, 2、3, 4两个数据子集，两个数据子集交由不同的同步传输单元完成。CDP数据同步面向是结构化、半结构化的数据同步，CDP对于源宿两端数据源均会抽象为一张二维表进行传输 和映射。CDP对于字段的映射是按照配置字段顺序进行一一映射，而非依赖源宿两端字段名称。CDP本身不对源宿两端字段做任何约束，视用户填写字段顺序一一传递。由于存在异构源宿两端字段个数可能

5、 不一致情况，CDP做如下约定：源端字段和目标端字段个数一致 CDP推荐配置源宿两端数据字段个数一致，避免数据字段补空情况。源端字段比目标端字段多源端字段比目标端字段列数多意味着一定存在数据丢失情况，为避免数据丢失，CDP对于这类配置直接报错。源端字段比目标端字段少 CDP对该类情况直接在导入数据末尾添加null，做补空处理。为做到云上所有数据存储能够做到四通八达的传输通道，CDP必须设计出一套通用的数据传输格式，同时还需 要保证各类数据存储系统均能够支持该协议。为尽可能适配最大范围系统，CDP仅设计支持常见数据类型，包PB 级云数据库 PetaData/服务条款括：CDP支持类型备注与说明L

6、ong定点数 Int、Short、Long、BigIntegerDouble浮点数 Float、Double、BigDecimal(无限精度)String字符串类型，底层不限长，使用通用字符集(Unicode)Date时间类型，包括日期、时间、日期时间类型Bool布尔值Bytes二进制，可以存放诸如MP3等非结构化数据不同的读取、写入插件负责将上述的表格数据转换到各自的数据存储系统。以Mysql读取为例：CDP支持类型Mysql插件对应类型Longint, tinyint, smallint, mediumint, int, bigintDoublefloat, double, decimal

7、Stringvarchar, char, tinytext, text, mediumtext, longtext, yearDatedate, datetime, timestamp, timeBoolbit, boolBytestinyblob, mediumblob, blob, longblob, varbinary同样，以ODPS写入的数据转换表如下：CDP支持类型ODPS插件对应类型LongBIGINTDoubleDOUBLEStringSTRINGDateDATETIMEBoolBOOLEANBytes无Mysql 所有int类型在ODPS最佳实践是存放为BIGINT类型，从上述

8、CDP两张类型转换对应表，CDP的转换也是符合该类类型设计。尽管CDP在数据类型通用性已经做最大可能适配，但仍请用户在使用CDP具体插件请查阅 每个插件对应的数据源类型，以确保CDP对插件类型支持符合您的预期。由于历史遗留问题，CDP在面对各类异构数据同步过程仍然难以避免源宿两端数据类型不匹配情况，例如用户密钥管理服务/API 文档需要从Mysql的Varchar类型导入ODPS的BIGINT类型。对于较为通用的数据类型转换CDP自身提供了统一的框 架级定义，方便用户需要在不同数据类型中做转换。类型 转换LongDoubleStringDateBoolByteLongYYY12NDoubleY

9、YYNNNString34Y56NDate7N8YNNBool9NYNYNBytesNN10NNYCDP默认的类型转换规则处理了大部分通用的数据转换规则。针对部分数据类型转换，CDP规则需要用户知晓：Long转换为Date，将Long数值视作Java提供Unix时间戳转换为Date，精度为毫秒。Long转换为Bool，0视为false，否则为trueString转换为Long，整数型字符串支持转换为Long，否则报错。不支持科学计数法字符串。String转换为Double，浮点型字符串支持转换为Long，否则报错。不支持科学计数法。String转换为Date，用户自定义时间格式(例如YYYY-

10、MM-DD)，满足用户时间格式可以转换，否则报错。String转换为Bool，true转为true，false转为false，其他报错。Date转为Long，将Date转为Unix时间戳。Date转换为String，根据用户自定义时间格式(例如YYYY-MM-DD)转换为字符串。Bool转换为Long，true转为1，false转为0。Bytes转换为String，根据用户指定的字符集编码转为String。注意:CDP传输过程中自动进行编码转换，例如CDP自动从Mysql GBK的字段导入ODPS UTF-8字段，用户不需要干预。CDP定位在为各类数据存储提供数据传输通道功能，在定期自动化运行

11、数据同步场景，如数仓ETL流程，要求所 有的数据同步Job能够做到多次数据同步能够和单次数据同步最终结果一致。例如当一次数据同步Job出现Fail情况，用户可以直接重启CDP任务而不需要到目的数据端进行线上数据清理操作。这在数仓领域属于Job幂 等性要求。CDP Job的幂等性是通过Writer插件的前置条件来实现的，例如用户可写ODPS配置中提供数据写入前的清理工作，该配置可以保证每次数据导入前都会先清除当前表或者分区的现有数据，这样能够保证数据多次写入的 结果和当前一次性写入结果一致。CDP对接云上各类数据存储系统，受限于各类数据存储模型事务支持(例如海量数据回流RDS是无法提供原子性归档

12、存储/API使用手册保证)，CDP本身无法提供写入的原子性语义保证：CDP Job数据同步过程中，一旦出现数据同步异常导致失败，CDP无法保证数据全部进入或者全部回滚(All orNothing)，同时数据导入过程对于目的数据源同样不是原子性提交，目的数据端仍可能看到数据进入的过程。对于原子语义，建议用户可以使用在系统工程中使用临时表等技巧绕过原子性语义问题。例如，在使用CDP将ODPS系统数据回流到生产RDS数据库，前端业务用户可能观察到海量数据导入RDS过程，包括部分数据进入RDS过程。对于该类应用，建议用户使用A、B表切换保证在线系统原子语义：当前在线服务使用A表完成OLTP操作，数仓回

13、流通道写入生产B表。待CDP整个写入RDS完成后，在线应用完成数据源从A表切换到B表，即可做到回流原子语义。在数据同步系统，可能由于最终客户人员误操作，或者由于分布式系统调用超时重试，有可能存在同一个Job被 多次并行启动。对于数据同步系统而言，数据多次重复调用极有可能造成数据重复写入，并引发数据故障。CDP本身提供了Job唯一性校验功能，例如同一份Job配置CDP客户同一时间启动两次及以上，CDP将自动识别 唯一性检查，对于重复提交Job，CDP将报错提示用户出现了Job多次提交。增量数据同步是数据同步业务层面实现，具体到底层数据同步分为批量Snapshot同步以及变更操作日志同步。CDP

14、Job通过数据存储数据API接口完成批量数据同步功能，因此CDP Job本身不支持获取数据存储系统层面的变化过程。例如CDP无法通过获取解析RDS BinLog日志反查数据变化量并同步到目的数据源。用户的增量业务需求，不仅仅只能通过系统级别增量变化日志解析实现数据增量同步。实际上，业界对于增量 同步有诸多解决方案：时间戳方式：它是一种基于快照比较的变化数据捕获方式，在源表上增加一个时间戳字段，系统中更 新修改表数据的时候，同时修改时间戳字段的值。当进行数据抽取时，通过比较上次抽取时间与时间 戳字段的值来决定抽取哪些数据。有的数据库的时间戳支持自动更新，即表的其它字段的数据发生改 变时，自动更新

15、时间戳字段的值。有的数据库不支持时间戳的自动更新，这就要求业务系统在更新业 务数据时，手工更新时间戳字段。在阿里数据库使用实践中，所有的在线业务系统按惯例都默认添加 了create_time、modify_time，对于在线业务写入均自动修改上述两个字段，记录修改时间。CDP在 进行数据同步过程中，仅需要在Mysql抽取环节配置where过滤条件做增量抽取。触发器方式：抽取的表上建立需要的触发器，一般要建立插入、修改、删除三个触发器，每当源表中 的数据发生变化，就被相应的触发器将变化的数据写入一个临时表，抽取线程从临时表中抽取数据，临时表中抽取过的数据被标记或删除。对于此类应用方式，CDP在进行数据同步时仅需指定抽取触 发器写入表即可。表方式：在业务系统中添加系统日志表，当业务数据发生变化时，更新维护日志表内容，当作ETL 加载时，通过读日志表数据决定加载那些数据及如何加载。对于此类应用方式，CDP同样使用Where过 滤条件做增量抽取。日志表方式：在业务系统中添加系统日志表，当业务数据发生变化时，更新维护日志表内容，当作ETL 加载时，通过读日志表数据决定加载那些数据及如何加载。对于此类应用方式，CDP同样使用Where过滤条件做增量抽取。数据同步通常会对接源宿两端数据存储，需要根据源宿两端数据源的具体信息适配和转换相应的数据内容。