暂无图片
暂无图片
暂无图片
暂无图片
暂无图片
首页 / ShardingSphere 数据迁移功能 & 实战

ShardingSphere 数据迁移功能 & 实战

ShardingSphere官微 2023-06-30
1064
📢 随着业务持续发展，数据量和并发量达到一定程度，传统数据库可能面临性能、可扩展性、可用性等问题。Apache ShardingSphere 是一款分布式的数据库生态系统， 可以将任意数据库转换为分布式数据库，并通过数据分片、弹性伸缩、加密等能力对原有数据库进行增强。本篇就为大家讲解 ShardingSphere 的数据迁移功能，并通过实战讲解搭建分布式数据库的整个流程。


数据迁移功能介绍

数据迁移内置于 ShardingSphere-Proxy，无需引入其他依赖。

数据迁移支持范围
目前支持的数据库类型
  • MySQL
  • PostgreSQL
  • openGauss
除此之外，如果新的数据库支持以上协议，那也在数据迁移支持范围内，支持这些数据库的同构及异构迁移。

数据迁移支持的数据库版本及配置示例：
数据库版本支持环境要求权限要求
MySQL5.1.15 ~ 8.0.xmy.cnf  配置
log-bin=binlog
binlog-format=row
binlog-row-image=full		GRANT REPLICATION SLAVE,REPLICATION CLIENT ON .TO ${usernamet}@${host}
PostgreSQL9.4 以上版本postgresql.conf  配置
wal_level = logical
max_wal_senders = 10
max_replication_slots = 10
max_connections = 600		pg_hba.conf  配置
host all ${username} 0.0.0.0/0 md5
openGauss2.0.1 ~ 3.1.xpostgresgl.conf  配置
wal_level = logical
max_wal_senders = 10
max_replication_slots = 10
max_connections = 600
wal_sender_timeout = 0		pg_hba.conf  配置
host all ${username} 0.0.0.0/0 md5
目前支持 MySQL 所有的主键类型，包括无主键。PostgreSQL 和 openGauss 由于增量插件暂时无法解析无主键类型表的部分事件，所以要求表具有主键或唯一键。

数据迁移相关概念

迁移的主要几个概念如下，后文中会提到这些概念。

数据迁移流程

上图是迁移的整个流程，ShardingSphere-Proxy 会读取源端的全量/增量数据，并写入到目标端。写入数据到目标端的时候，会完成数据分片、加密/脱敏等操作，是否需要这些能力可以根据需求自定义。

数据迁移阶段

数据迁移主要分为 4 个阶段 ：
1、准备阶段 
在准备阶段，数据迁移模块会进行数据源连通性及权限的校验，同时进行存量数据的统计、日志位点的记录，进行任务的初始化 
2、存量迁移阶段 
存量迁移阶段采用 JDBC 查询的方式，直接从源端读取数据，基于配置的分片等规则写入到目标端。 
3、增量迁移阶段 
由于存量数据迁移耗费的时间受到数据量和并行度等因素影响，需要对这段时间内业务新增的数据进行同步。MySQL是通过订阅并解析 binlog 实现的，当增量数据基本同步完成时（由于业务系统未停止，增量数据是不断的），则进入流量切换阶段。 
4、流量切换阶段 
迁移完成后，用户可以把读流量或者写流量切换到 Apache ShardingSphere。

用户可以通过相应的 DistSQL 实时查看迁移进度。同时可以通过数据一致性校验功能，对迁移前后的数据进行比对。


搭建分布式数据库实战

环境准备

  1. ShardingSphere 版本
5.4.0 or master branch & cluster mode
  1. MySQL数据库
源端 1 台，目标端 2 台（一主一从），共计 3 台数据库
  1. 注册中心
ZooKeeper 3.8.0

部署架构

其中 ShardingSphere-Proxy 作为计算节点，MySQL 作为存储节点，用到了数据分片、读写分离、加密等能力对原有数据库进行增强。
通过数据迁移功能，将 source_ds 中的数据迁移到 target 集群中。

数据库权限配置

在业务中，一般不会使用管理员账户进行数据迁移，所以需要先对源端数据库配置迁移所需要的权限。
类似的，由于迁移会涉及到自动建表、建索引，所以如果目标端使用的是普通账户，需要提前赋予相应的权限
GRANT CREATE, ALTER, DROP, SELECT, INSERT, UPDATE, DELETE, INDEX ON target_ds_0.* TO target_user; 
GRANT CREATE, ALTER, DROP, SELECT, INSERT, UPDATE, DELETE, INDEX ON target_ds_1.* TO target_user;



操作步骤

初始化源端数据

首先初始化源端的表
CREATE TABLE t_user
(
    id       int auto_increment,
    username varchar(64)  null,
    password varchar(255) null,
    mobile   varchar(64)  null,
    constraint t_user_pk primary key (id)
);
模拟初始数据
INSERT INTO t_user (id, username, password, mobile) VALUES (1, 'jack', '123456', '13111111111');
INSERT INTO t_user (id, username, password, mobile) VALUES (2, 'rose', '234567', '13111111112');
INSERT INTO t_user (id, username, password, mobile) VALUES (3, 'mike', 'aaa123', '13111111113');
INSERT INTO t_user (id, username, password, mobile) VALUES (4, 'bob', 'aaabbb', '13111111114');

初始化 ShardingSphere Proxy 规则

首先使用集群模式启动 ShardingSphere Proxy，并通过数据库客户端登录，执行创建数据库的命令
mysql> create database sharding_db;
Query OK, 0 rows affected (0.12 sec)

mysql> use sharding_db
Database changed
  1. 使用 DistSQL 添加存储单元
REGISTER STORAGE UNIT target_ds_0  (
     URL="jdbc:mysql://localhost:3306/target_ds_0?serverTimezone=UTC&useSSL=false&allowPublicKeyRetrieval=true",
     USER="target_user",
     PASSWORD="root"
);
REGISTER STORAGE UNIT target_ds_1  (
     URL="jdbc:mysql://localhost:3306/target_ds_1?serverTimezone=UTC&useSSL=false&allowPublicKeyRetrieval=true",
     USER="target_user",
     PASSWORD="root" 
);
REGISTER STORAGE UNIT read_ds_0  (
     URL="jdbc:mysql://localhost:3308/target_ds_0?serverTimezone=UTC&useSSL=false&allowPublicKeyRetrieval=true",     
     USER="target_user",
     PASSWORD="root"
);
REGISTER STORAGE UNIT read_ds_1  (
     URL="jdbc:mysql://localhost:3308/target_ds_1?serverTimezone=UTC&useSSL=false&allowPublicKeyRetrieval=true",     
     USER="target_user",
     PASSWORD="root"
);

  1. 初始化规则定义
初始化读写分离规则，后续迁移使用的是这里的逻辑数据源
CREATE READWRITE_SPLITTING RULE rw_ds_0 (
    WRITE_STORAGE_UNIT=target_ds_0,
    READ_STORAGE_UNITS(read_ds_0),
    TYPE(NAME="random")
);
CREATE READWRITE_SPLITTING RULE rw_ds_1 (
    WRITE_STORAGE_UNIT=target_ds_1,
    READ_STORAGE_UNITS(read_ds_1),
    TYPE(NAME="random")
);
初始化分片规则
CREATE SHARDING TABLE RULE t_user(
        STORAGE_UNITS(rw_ds_0, rw_ds_1),
        SHARDING_COLUMN=id,
        TYPE(NAME="hash_mod",PROPERTIES("sharding-count"="4")),
        KEY_GENERATE_STRATEGY(COLUMN=id,TYPE(NAME="snowflake"))
);

初始化加密规则
CREATE ENCRYPT RULE t_user (
    COLUMNS((NAME=password,CIPHER=password_cipher,ENCRYPT_ALGORITHM(TYPE(NAME='AES',PROPERTIES('aes-key-value'='123456abc'))))
));

至此，目标端相关的配置已经初始化完毕。

开始数据迁移

1. 添加迁移数据源

迁移的数据源是外部的，需要通过 DistSQL 注册

REGISTER MIGRATION SOURCE STORAGE UNIT source_ds (
    URL="jdbc:mysql://${source_database_url:port}/source_ds?serverTimezone=UTC&useSSL=false",
    USER="root",
    PASSWORD="root"
);

  1. 执行迁移命令
MIGRATE TABLE source_ds.t_user INTO sharding_db.t_user;

这里的source_ds就是上一步注册的源端数据源，sharding_db就是在 ShardingSphere-Proxy 中创建的数据库名。示例结果
mysql> MIGRATE TABLE source_ds.t_user INTO sharding_db.t_user;
Query OK, 0 rows affected (1.06 sec)

  1. 查看迁移进度
通过 SHOW MIGRATION LIST 查询迁移作业列表
mysql> SHOW MIGRATION LIST;
+--------------------------------------------+------------------+----------------+--------+---------------------+-----------+
| id                                         | tables           | job_item_count | active | create_time         | stop_time |
+--------------------------------------------+------------------+----------------+--------+---------------------+-----------+
| j0102p000041c4912117c302e9facd92f9a74a478c | source_ds.t_user | 1              | true   | 2023-06-24 09:44:51 | NULL      |
+--------------------------------------------+------------------+----------------+--------+---------------------+-----------+
1 row in set (0.03 sec)

返回的结果字段释义如下
  1. id：jobId，唯一标识
  2. tables：迁移的表名
  3. job_item_count：迁移 job 的分片数
  4. active：job 是否在运行中
  5. create_time：job 创建时间
  6. stop_time：job 停止时间

根据上面拿到的 jobId，可以通过 SHOW MIGRATION STATUS jobId 查询到作业的详情，示例如下
mysql> SHOW MIGRATION STATUS j0102p000041c4912117c302e9facd92f9a74a478c;
+------+-------------+------------------+--------------------------+--------+-------------------------+-------------------------------+--------------------------+---------------+
| item | data_source | tables           | status                   | active | processed_records_count | inventory_finished_percentage | incremental_idle_seconds | error_message |
+------+-------------+------------------+--------------------------+--------+-------------------------+-------------------------------+--------------------------+---------------+
| 0    | source_ds   | source_ds.t_user | EXECUTE_INCREMENTAL_TASK | true   | 4                       | 100                           | 321                      |               |
+------+-------------+------------------+--------------------------+--------+-------------------------+-------------------------------+--------------------------+---------------+
1 row in set (0.04 sec)

返回的结果字段释义如下
  1. item：job分片任务编号
  2. data_source：迁移数据源名称
  3. tables：迁移表名
  4. status：job状态，具体可以参考 JobStatus
  5. active：job 是否在运行中
  6. processed_records_count：已经处理的数据量
  7. inventory_finished_percentage：全量完成百分比
  8. incremental_idle_seconds：增量空闲时间
  9. error_message：错误信息

在迁移的过程中，全量阶段和增量阶段会衔接上，确保数据的完整性。在生成写入目标端的 INSERT SQL 时，会进 SQL 改写，确保同一条数据的插入操作是幂等的，由于注册中心的进度可能不是最新的，在出现重启job的时候，有可能出现重复消费的情况，所以需要保证幂等。

验证迁移前后的数据一致性

迁移前后的数据源是独立的，在数据量大的时候很难通过肉眼去验证数据的一致性，因此数据迁移功能还提供了数据一致性校验的功能。用户可以通过 DistSQL 对前后的数据进行比对。
由于增量数据同步存在延时，数据一致性校验需要如下前置条件：
  • 数据迁移进入到增量迁移的阶段
  • 需要一定时间的业务只读窗口期

示例如下：
mysql> CHECK MIGRATION j0102p000041c4912117c302e9facd92f9a74a478c BY TYPE (NAME='DATA_MATCH');
Query OK, 0 rows affected (0.48 sec)

mysql> SHOW MIGRATION CHECK STATUS j0102p000041c4912117c302e9facd92f9a74a478c;
+------------------+--------+---------------------+---------------------+-------------------+-------------------------+-------------------------+------------------+---------------+
| tables           | result | check_failed_tables | finished_percentage | remaining_seconds | check_begin_time        | check_end_time          | duration_seconds | error_message |
+------------------+--------+---------------------+---------------------+-------------------+-------------------------+-------------------------+------------------+---------------+
| source_ds.t_user | true   |                     | 100                 | 0                 | 2023-06-24 10:05:28.483 | 2023-06-24 10:05:29.821 | 1                |               |
+------------------+--------+---------------------+---------------------+-------------------+-------------------------+-------------------------+------------------+---------------+
1 row in set (0.06 sec)

字段释义如下：
  1. tables：表名
  2. result：校验结果，true 表示通过，false 表示存在不一致
  3. check_failed_tables：校验失败的表
  4. finished_percentage：进度
  5. remaining_seconds：剩余时间（估算）
  6. check_begin_time：校验开始时间
  7. check_end_time：校验结束时间

数据一致性校验算法可以通过 SHOW MIGRATION CHECK ALGORITHMS 进行查看，支持指定校验算法
mysql> SHOW MIGRATION CHECK ALGORITHMS;
+-------------+--------------------------------------------------------------+----------------------------+
| type        | supported_database_types                                     | description                |
+-------------+--------------------------------------------------------------+----------------------------+
| CRC32_MATCH | MySQL,MariaDB,H2                                             | Match CRC32 of records.    |
| DATA_MATCH  | SQL92,MySQL,MariaDB,PostgreSQL,openGauss,Oracle,SQLServer,H2 | Match raw data of records. |
+-------------+--------------------------------------------------------------+----------------------------+
2 rows in set (0.10 sec)

MySQL 支持 CRC32 校验算法和 DATA_MATCH 校验算法
  • CRC32_MATCH：循环冗余校验，通过校验码来判断是否存在数据不一致，效率快，但是不支持断点续传，且只支持MySQL
  • DATA_MATCH：逐行挨个比对数据，效率稍慢但是支持断点续传和异构数据库
目标端开启数据加密的情况需要使用 DATA_MATCH，因为底层的字段名称或者是值会存在不一致。异构迁移也需要使用 DATA_MATCH。


提交迁移作业

最后一步就是提交操作，在做提交操作之前，要求已经完成了数据一致性校验。该步骤会做一些收尾的操作，例如清理 PostgreSQL 的 Replication Slot，并且在最后刷新表元数据。
mysql> COMMIT MIGRATION j0102p000041c4912117c302e9facd92f9a74a478c;
Query OK, 0 rows affected (2.79 sec)

至此，迁移的整个流程结束。可以将流量切换为 ShardingSphere-Proxy 迁移完成的表。

总结

通过 ShardingSphere 提供的数据迁移方案可以助力传统数据库平滑切换到 ShardingSphere，以上就是本次分享的全部内容，如果读者对 Apache ShardingSphere 有任何疑问或建议，欢迎在 GitHub issue 列表 [2] 提出，或可前往中文社区 [3] 交流讨论。

🔗 参考

[1] ShardingSphere-数据迁移 官网文档：https://shardingsphere.apache.org/document/current/cn/user-manual/shardingsphere-proxy/migration/usage/
[2] GitHub issue 列表：https://github.com/apache/shardingsphere/issues
[3] 中文社区：https://community.sphere-ex.com/

mysqlshardingspheremysql创建数据库mysql读写分离数据迁移
文章转载自ShardingSphere官微，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

评论