大数据开发治理平台 DataWorks：DRDS（PolarDB-X 1.0）数据源

支持的版本

实时读取

支持DRDS（PolarDB-X 1.0）实例。创建DRDS（PolarDB-X 1.0）实例的操作可参见创建PolarDB-X 1.0实例。

使用限制

离线读写

• 目前DRDS（PolarDB-X 1.0）的插件仅适配MySQL引擎的场景。DRDS（PolarDB-X 1.0）是一套分布式MySQL数据库，并且大部分通信协议遵守MySQL使用场景。

• DRDS（PolarDB-X 1.0）下的MySQL8.0版本支持使用Serverless资源组（推荐）和独享数据集成资源组。

• 在底层实现上，DRDS（PolarDB-X 1.0） Writer通过JDBC连接远程DRDS（PolarDB-X 1.0）数据库的Proxy，执行相应的replace into语句，写入数据至DRDS（PolarDB-X 1.0）。

  执行的SQL语句是replace into，为避免数据重复写入，需要您的表具备主键（Primary Key）或唯一性索引（Unique index）。

• DRDS（PolarDB-X 1.0） Writer通过数据同步框架获取Reader生成的协议数据，通过replace into（没有遇到主键/唯一性索引冲突时，与insert into行为一致，冲突时会用新行替换原有行所有字段）语句写入数据至DRDS（PolarDB-X 1.0）。DRDS（PolarDB-X 1.0） Writer累积一定数据，提交给DRDS（PolarDB-X 1.0）的Proxy，该Proxy内部决定数据是写入一张还是多张表，以及多张表写入时如何路由数据。

  说明

  整个任务至少需要具备replace into的权限。是否需要其它权限，取决于您配置任务时在preSql和postSql中指定的语句。

• 支持读取视图表。

实时读

• 目前仅支持使用实例模式配置的数据源，如果您使用JDBC连接串配置的数据源，任务运行将会出错。

• 实例的存储类型：仅支持PolarDB（即租户侧PolarDB MySQL）和存量的用户RDS（新购已不支持），不支持RDS MySQL（即私有定制RDS MySQL）。

• 不支持XA ROLLBACK。

  针对已经XA PREPARE的事务数据，实时同步会将其同步到目标端，如果XA ROLLBACK，实时同步不会针对XA PREPARE的数据做回滚写入的操作。若要处理XA ROLLBACK场景，需要手动将XA ROLLBACK的表从实时同步任务中移除，再添加表后重新进行同步。

支持的字段类型

DRDS（PolarDB-X 1.0） Reader和DRDS（PolarDB-X 1.0） Writer支持大部分DRDS（PolarDB-X 1.0）类型，但也存在个别类型没有支持的情况，请注意检查您的数据类型。

DRDS（PolarDB-X 1.0） Reader和DRDS（PolarDB-X 1.0） Writer针对DRDS（PolarDB-X 1.0）类型的转换列表，如下所示。

数据同步前准备

您需要规划一个数据库的登录账户用于后续执行操作，此账号需要拥有DRDS（PolarDB-X 1.0）的相关操作权限，详情请参见新建账号。

创建数据源

在进行数据同步任务开发时，您需要在DataWorks上创建一个对应的数据源，操作流程请参见创建并管理数据源，详细的配置参数解释可在配置界面查看对应参数的文案提示。

数据同步任务开发

数据同步任务的配置入口和通用配置流程可参见下文的配置指导。

常见问题

• 一致性视图问题

  DRDS（PolarDB-X 1.0）本身属于分布式数据库，对外无法提供一致性的多库多表视图。不同于MySQL等单库单表同步，DRDS（PolarDB-X 1.0） Reader无法抽取同一个时间切片的分库分表快照信息，即DRDS（PolarDB-X 1.0） Reader抽取底层不同的分表将获取不同的分表快照，无法保证强一致性。

• 数据库编码问题

  DRDS（PolarDB-X 1.0）本身的编码设置非常灵活，包括指定编码到库、表、字段级别，甚至可以设置不同编码。优先级从高到低为字段、表、库、实例。建议您在库级别将编码统一设置为UTF-8。

  DRDS（PolarDB-X 1.0） Reader底层使用JDBC进行数据抽取，JDBC天然适配各类编码，并在底层进行了编码转换。因此DRDS（PolarDB-X 1.0） Reader不需要您指定编码，可以自动获取编码并转码。

  对于DRDS（PolarDB-X 1.0）底层写入编码和其设定的编码不一致的混乱情况，DRDS（PolarDB-X 1.0） Reader对此无法识别，该类情况的同步结果有可能为乱码。

• 增量数据同步的方式

  DRDS（PolarDB-X 1.0） Reader使用JDBC SELECT语句完成数据抽取工作，因此您可以使用SELECT…WHERE…进行增量数据抽取，方式如下：

  • 数据库在线应用写入数据库时，填充modify字段为更改时间戳，包括新增、更新、删除（逻辑删除）。对于这类应用，DRDS（PolarDB-X 1.0） Reader只需要where条件后跟上一同步阶段时间戳即可。

  • 对于新增流水型数据，DRDS（PolarDB-X 1.0） Reader在where条件后跟上一阶段最大自增ID即可。

  对于业务上无字段区分新增、修改数据的情况，DRDS（PolarDB-X 1.0） Reader无法进行增量数据同步，只能同步全量数据。

• 不支持在where语句中配置物理表相关的筛选条件。

附录：脚本Demo与参数说明

{
    "type":"job",
    "version":"2.0",//版本号
    "steps":[
        {
            "stepType":"drds",//插件名
            "parameter":{
                "datasource":"",//数据源名
                "column":[//列名
                    "id",
                    "name"
                ],
                "where":"",//过滤条件
                "table":"",//表名
                "splitPk": ""//切分键
            },
            "name":"Reader",
            "category":"reader"
        },
        {
            "stepType":"stream",//插件名
            "parameter":{},
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"//错误记录数
        },
        "speed":{
            "throttle":true,//当throttle值为false时，mbps参数不生效，表示不限流；当throttle值为true时,表示限流。
            "concurrent":1,//并发数
            "mbps":"12"//限流，此处1mbps = 1MB/s。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

{
    "type":"job",
    "version":"2.0",//版本号。
    "steps":[
        {
            "stepType":"stream",
            "parameter":{},
            "name":"Reader",
            "category":"reader"
                },
        {
            "stepType":"drds",//插件名。
            "parameter":{
                "postSql":[],//执行数据同步任务之后执行的SQL语句。
                "datasource":"",//数据源。
                "column":[//列名。
                "id"
                ],
                "writeMode":"insert ignore",
                "batchSize":"1024",//一次性批量提交的记录数大小。
                "table":"test",//表名。
                "preSql":[]//执行数据同步任务之前执行的SQL语句。
                },
            "name":"Writer",
            "category":"writer"
                }
                ],
    "setting":{
        "errorLimit":{
        "record":"0"//错误记录数。
            },
        "speed":{
            "throttle":true,//当throttle值为false时，mbps参数不生效，表示不限流；当throttle值为true时,表示限流。
            "concurrent":1, //作业并发数。
            "mbps":"12"//限流，此处1mbps = 1MB/s。
                }
            },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
                }
            ]
        }
    }