MySQL数据源

实时读

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

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

• 不支持XA ROLLBACK。

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

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

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

• 对于Amazon Aurora MySQL数据库，需要连接到您的主/写数据库，因为AWS不允许在Aurora MySQL的只读副本上激活Binlog功能。实时同步任务需要Binlog来执行增量更新。

• 实时同步在线DDL变更仅支持通过数据管理DMS对MySQL表进行加列（Add Column）在线DDL变更。

• 不支持读取MySQL中的存储过程。

离线读

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

• 不支持读取MySQL中的存储过程。

确认MySQL版本

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

SELECT version();

配置账号权限

建议您提前规划并创建一个专用于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语句为全局权限，不能指定授权同步账号对目标数据库的指定表拥有相关权限。

开启MySQL Binlog（仅实时同步需要）

数据集成通过实时订阅MySQL Binlog实现增量数据实时同步，您需要在DataWorks配置同步前，先开启MySQL 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请参考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。

3. 查询Binlog完整日志是否开启。

   使用如下语句查询Binlog完整日志是否开启。

   SHOW variables LIKE "binlog_row_image";

   返回结果说明：

   • 返回FULL，表示Binlog开启了完整日志。

   • 返回MINIMAL，表示Binlog开启了最小日志，未开启完整日志。

   重要

   DataWorks实时同步，仅支持同步开启了Binlog完整日志的MySQL服务器数据。若查询结果返回非FULL，请修改binlog_row_image的配置。

阿里云实例模式

若您的MySQL为阿里云RDS实例，建议您选择阿里云实例模式创建数据源。参数配置方式如下：

连接串模式

你也可以选择连接串模式创建数据源，方式更为灵活。连接串模式的参数配置方式如下：

离线任务脚本配置方式

如果您配置离线任务时使用脚本模式的方式进行配置，您需要按照统一的脚本格式要求，在任务脚本中编写相应的参数，详情请参见脚本模式配置，以下为您介绍脚本模式下数据源的参数配置详情。

Reader脚本Demo

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

• 配置单库单表

  {
    "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;"
              ],
              "datasource": ""//数据源名称。
            }
          ],
          "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"
        }
      ]
    }
  }

Reader脚本参数

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": "",//表名。
        "nullMode": "skipNull",//NULL值处理策略。
        "skipNullColumn": [//需要跳过NULL值的列。
          "id",
          "value"
        ],
        "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"
      }
    ]
  }
}

Writer脚本参数