阿里云大数据开发治理平台 DataWorksMySQL数据源-云淘科技

支持的MySQL版本

• 离线读写：

  支持MySQL 5.5.x、MySQL 5.6.x、MySQL 5.7.x、MySQL 8.0.x，兼容Amazon RDS for MySQL。

  离线同步支持读取视图表。

• 实时读取：

  数据集成实时读取MySQL数据是基于实时订阅MySQL实现的，当前仅支持实时同步MySQL 5.5.x、MySQL 5.6.x、MySQL 5.7.x、MySQL 8.0.x（非8.0新特性，比如 functional index，仅兼容原有功能）版本的MySQL数据，兼容Amazon RDS for MySQL。

  重要

  如果需要同步DRDS的MySQL，请不要将DRDS的MySQL配置为MySQL数据源，您可以参考配置DRDS数据源文档直接将其配置为DRDS数据源。

使用限制

实时读

• 不支持同步MySQL只读库实例的数据。

• 不支持同步含有Functional index的表。

• 不支持XA ROLLBACK。

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

• 仅支持同步MySQL服务器Binlog配置格式为ROW。

• 实时同步不会同步被级联删除的关联表记录。

离线读

MySQL Reader插件在进行分库分表等多表同步时，若要对单表进行切分，则需要满足任务并发数大于表个数这一条件，否则切分的Task数目等于表的个数。

支持的字段类型

各版本MySQL的全量字段类型请参见MySQL官方文档。以下以MySQL 8.0.x为例，为您罗列当前主要字段的支持情况。

数据同步前准备：MySQL环境准备

在DataWorks上进行数据同步前，您需要参考本文提前在MySQL侧进行数据同步环境准备，以便在DataWorks上进行MySQL数据同步任务配置与执行时服务正常。以下为您介绍MySQL同步前的相关环境准备。

准备工作1：确认MySQL版本

数据集成对MySQL版本有要求，您可参考上文支持的MySQL版本章节，查看当前待同步的MySQL是否符合版本要求。您可以在MySQL数据库通过如下语句查看当前MySQL数据库版本。

select version();

准备工作2：配置账号权限

建议您提前规划并创建一个专用于DataWorks访问数据源的MySQL账号，操作如下。

1. 可选：创建账号。

   操作详情请参见创建MySQL账号。

2. 配置权限。

   • 离线

     在离线同步场景下：

     • 在离线读MySQL数据时，此账号需拥有同步表的读（SELECT）权限。

     • 在离线写MySQL数据时，此账号需拥有同步表的写（INSERT、DELETE、UPDATE）权限。

   • 实时

     在实时同步场景下，此账号需要拥有数据库的SELECT、REPLICATION SLAVE、REPLICATION CLIENT权限。

   您可以参考以下命令为账号添加权限，或直接给账号赋予SUPER权限。如下执行语句在实际使用时，请替换'同步账号'为上述创建的账号。

   -- CREATE USER '同步账号'@'%' IDENTIFIED BY '密码'; //创建同步账号并设置密码，使其可以通过任意主机登录数据库。%表示任意主机。
   GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO '同步账号'@'%'; //授权同步账号数据库的 SELECT, REPLICATION SLAVE, REPLICATION CLIENT权限。

   *.*表示授权同步账号对所有数据库的所有表拥有上述权限。您也可以指定授权同步账号对目标数据库的指定表拥有上述权限。例如，授权同步账号对test数据库的user表拥有上述权限，则可以使用GRANT SELECT, REPLICATION CLIENT ON test.user TO '同步账号'@'%';语句。

   说明

   REPLICATION SLAVE语句为全局权限，不能指定授权同步账号对目标数据库的指定表拥有相关权限。

准备工作3：（仅实时同步需要）开启MySQL Binlog

数据集成通过实时订阅MySQL Binlog实现增量数据实时同步，您需要在DataWorks配置同步前，先开启MySQL Binlog服务。操作如下。

重要

• 如果Binlog在消费中，则无法被数据库删除。如果实时同步任务运行延迟将可能导致源端Binlog长时间被消费，请合理配置任务的延迟告警，并及时关注数据库的磁盘空间。

• Binlog至少保留72小时以上，避免任务失败后因Binlog已经消失，再启动无法重置位点到故障发生前而导致的数据丢失（此时只能使用全量离线同步来补齐数据）。

1. 检查Binlog是否开启。

   • 使用如下语句检查Binlog是否开启。

     show variables like "log_bin";

     返回结果为ON时，表明已开启Binlog。

   • 如果您使用备用库同步数据，则还可以通过如下语句检查Binlog是否开启。

     show variables like "log_slave_updates";

     返回结果为ON时，表明备用库已开启Binlog。

   如果返回的结果与上述结果不符：

   • 开源MySQL请参考MySQL官方文档开启Binlog。

   • 阿里云RDS MySQL请参考日志备份开启Binlog。

   • 阿里云PolarDB MySQL请参考开启Binlog开启Binlog。

2. 查询Binlog的使用格式。

   使用如下语句查询Binlog的使用格式。

   show variables like "binlog_format";

   返回结果说明：

   • 返回ROW，表明开启的Binlog格式为ROW。

   • 返回STATEMENT，表明开启的Binlog格式为STATEMENT。

   • 返回MIXED，表明开启的Binlog格式为MIXED。

   重要

   DataWorks实时同步仅支持同步MySQL服务器Binlog配置格式为ROW。如果返回非ROW请修改Binlog Format。

数据同步任务开发：MySQL同步流程引导

MySQL数据同步任务的配置入口和通用配置流程指导可参见下文的配置指导，详细的配置参数解释可在配置界面查看对应参数的文案提示。

创建数据源

在进行数据同步任务开发时，您需要在DataWorks上创建一个对应的数据源，操作流程请参见创建与管理数据源。

单表离线同步任务配置指导

• 操作流程请参见通过向导模式配置离线同步任务、通过脚本模式配置离线同步任务。

• 脚本模式配置的全量参数和脚本Demo请参见下文的附录：MySQL脚本Demo与参数说明。

单表实时同步任务配置指导

操作流程请参见DataStudio侧实时同步任务配置。

整库离线、整库（实时）全增量、整库（实时）分库分表等整库级别同步配置指导

操作流程请参见数据集成侧同步任务配置。

常见问题

• 实时同步MySQL数据源的数据时，一开始读到数据，一段时间后无法读到数据，怎么处理？

• 实时同步MySQL数据报错

• 实时同步MySQL数据源时速度为什么会变慢？

• MySQL分库分表如何将分表同步到一张MaxCompute中

• 目的端MySQL表字符集为utf8mb4时，同步到MySQL中的中文字符出现乱码时，如何处理？

• 如何处理编码格式设置/乱码问题导致的脏数据报错？

• 如何排查离线同步任务运行时间长的问题？

• 写入/读取MySQL报错：Application was streaming results when the connection failed. Consider raising value of ‘net_write_timeout/net_read_timeout、’ on the server.

更多其他数据集成常见问题请参见数据集成常见问题。

附录：MySQL脚本Demo与参数说明

附录：离线任务脚本配置方式

如果您配置离线任务时使用脚本模式的方式进行配置，您需要在任务脚本中按照脚本的统一格式要求编写脚本中的reader参数和writer参数，脚本模式的统一要求请参见通过脚本模式配置离线同步任务，以下为您介绍脚本模式下的数据源的Reader参数和Writer参数的指导详情。

MySQL Reader脚本Demo

本文为您提供单库单表和分库分表的配置示例：

• 配置单库单表

  querySql{
      "type":"job",
      "version":"2.0",//版本号。
      "steps":[
          {
              "stepType":"mysql",//插件名。
              "parameter":{
                  "column":[//列名。
                      "id"
                  ],
                  "connection":[
                      {   "querySql":"select a,b from join1 c join join2 d on c.id = d.id;", //使用字符串的形式，将querySql写在connection中。
                          "datasource":"",//数据源。
                          "table":[//表名，即使只有一张表，也必须以[]的数组形式书写。
                              "xxx"
                          ]
                      }
                  ],
                  "where":"",//过滤条件。
                  "splitPk":"",//切分键。
                  "encoding":"UTF-8"//编码格式。
              },
              "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"
              }
          ]
      }
  }

• 配置分库分表

  说明

  分库分表是指在MySQL Reader端可以选择多个MySQL数据表，且表结构保持一致。此处的‘分库分表’是指多个MySQL写入同一个目标表，如想要支持整库级别配置分库分表，还请在数据集成站点建立任务并选择整库分库分表能力

  {
      "type": "job",
      "version": "2.0",
      "steps": [
          {
              "stepType": "mysql",
              "parameter": {
                  "indexes": [
                      {
                          "type": "unique",
                          "column": [
                              "id"
                          ]
                      }
                  ],
                  "envType": 0,
                  "useSpecialSecret": false,
                  "column": [
                      "id",
                      "buyer_name",
                      "seller_name",
                      "item_id",
                      "city",
                      "zone"
                  ],
                  "tableComment": "测试订单表",
                  "connection": [
                      {
                          "datasource": "rds_dataservice",
                          "table": [
                              "rds_table"
                          ]
                      },
                      {
                          "datasource": "rds_workshop_log",
                          "table": [
                              "rds_table"
                          ]
                      }
                  ],
                  "where": "",
                  "splitPk": "id",
                  "encoding": "UTF-8"
              },
              "name": "Reader",
              "category": "reader"
          },
          {
              "stepType": "odps",
              "parameter": {
                             },
              "name": "Writer",
              "category": "writer"
          },
          {
              "name": "Processor",
              "stepType": null,
              "category": "processor",
              "copies": 1,
              "parameter": {
                  "nodes": [],
                  "edges": [],
                  "groups": [],
                  "version": "2.0"
              }
          }
      ],
      "setting": {
          "executeMode": null,
          "errorLimit": {
              "record": ""
          },
          "speed": {
              "concurrent": 2,
              "throttle": false
          }
      },
      "order": {
          "hops": [
              {
                  "from": "Reader",
                  "to": "Writer"
              }
          ]
      }
  }

MySQL Reader脚本参数

MySQL writer脚本Demo

{
    "type":"job",
    "version":"2.0",//版本号。
    "steps":[ 
        {
            "stepType":"stream",
            "parameter":{},
            "name":"Reader",
            "category":"reader"
        },
        {
            "stepType":"mysql",//插件名。
            "parameter":{
                "postSql":[],//导入后的准备语句。
                "datasource":"",//数据源。
                "column":[//列名。
                    "id",
                    "value"
                ],
                "writeMode":"insert",//写入模式，您可以设置为insert、replace或update。
                "batchSize":1024,//一次性批量提交的记录数大小。
                "table":"",//表名。
                "preSql":[ 
                     "delete from XXX;" //导入前的准备语句。
                   ]
            },
            "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"
            }
        ]
    }
}

MySQL Writer脚本参数