使用DTS同步MySQL增量数据到Tablestore-阿里云开发者社区

开发者社区> 阿里云存储服务> 正文

使用DTS同步MySQL增量数据到Tablestore

简介: 摘要 从MySQL到Tablestore的全量数据导出可以参考同系列文章《数据同步-从MySQL到Tablestore》,本文主要介绍将MySQL的增量数据同步到Tablestore的一种方式——使用阿里集团的数据传输服务DTS的数据订阅功能做增量数据的读取以及改写。

摘要

从MySQL到Tablestore的全量数据导出可以参考同系列文章《数据同步-从MySQL到Tablestore》,本文主要介绍将MySQL的增量数据同步到Tablestore的一种方式——使用阿里集团的数据传输服务DTS的数据订阅功能做增量数据的读取以及改写。
注意:DTS数据订阅服务支持多种数据库环境,老版现不支持MySQL8.0,使用sdk进行消费;新版新增了分组消费概念,需要使用Kafka客户端消费订阅数据。本文以RDS(MySQL 5.7)订阅为例,使用sdk完成增量数据订阅与改写。

原理介绍

_2_

导出步骤

1.源、目的数据库资源

源数据库:

RDS(新建实例)/实例[pingsheng]/数据库[pingstest]/表[to_tablestore]
数据表结构如图
_00002

目的数据库:

Tablestore(新建实例)/实例[pingsheng]/表[from_rds]
数据表结构如图
_00010

2.云账号资源

准备具有源、目的数据库读写权限的一组云账号AK

3.DTS数据订阅

创建订阅通道参考,选择上述源数据库实例为数据源配置订阅信息
_001
选择需要订阅的数据表
_003
通过数据源预检查后,数据订阅配置完成,进入初始化阶段大约需要等待十分钟。初始化完成后,数据订阅状态变为“正常”即可以开始消费增量数据。增量数据的消费点从界面可以看到,支持动态调整参考文档
_004
从控制台的“订阅数据”可以看到已经拉取到的部分展示数据
_005
从DTS拉取到的增量数据是经过解析和再封装的,增添了一些解释参数,订阅数据的各字段含义参考

4.订阅数据的解析与改写

从DTS读取MySQL增量数据

下载DTS的SDK,在本地(ECS)进行编译,参考
_006
在数据订阅“更多”中下载示例代码,替换掉AK信息、订阅ID,编译启动程序尝试获取增量数据,测试rds数据表中若无增量,会每隔1s收到一条“heartbeat”心跳记录
_007
尝试在源数据表insert、update数据,会打印出以Opt:begin开头,包含Opt:insert、update,以Opt:commit结尾的多行数据。修改代码仅保留改写数据需要的操作类型“Opt”和行信息的前后镜像“FieldList”

public void notify(List<ClusterMessage> messages) throws Exception {
  for (ClusterMessage message : messages) {
    // debug
    System.out.println(message.getRecord().getOpt());
    System.out.println(message.getRecord().getFieldList());
    //you must call ackAsConsumed when you consume the data
    message.ackAsConsumed();
  }
}

//BEGIN
//[]
//UPDATE
//[Field name: pk1  //依次输出各列的前、后镜像
//Field type: 3
//Field length: 2
//Field value: 83
//,Field name: pk1
//Field type: 3
//Field length: 2
//Field value: 80
//, Field name: pk2
//Field type: 3
//Field length: 1
//Field value: 3
//, Field name: pk2
//Field type: 3
//Field length: 1
//Field value: 3
//, Field name: v1
//Field type: 3
//Field length: 2
//Field value: 47
//, Field name: v1
//Field type: 3
//Field length: 2
//Field value: 50
//]
//COMMIT
//[]

将增量数据写入Tablestore

下载Tablestore的SDK ,本地(ECS)进行编译参考
调用单行数据操作,将增、删、改的行写入Tablestore参考

//PutRow
private static void putRow(SyncClient client, String pkValue, MyColumnValue columnvalue) {
    // 构造主键
    PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
    primaryKeyBuilder.addPrimaryKeyColumn(PRIMARY_KEY_NAME, PrimaryKeyValue.fromString(pkValue));
    PrimaryKey primaryKey = primaryKeyBuilder.build();
    RowPutChange rowPutChange = new RowPutChange(TABLE_NAME, primaryKey);
    //加入属性列
    rowPutChange.addColumn(new Column("v1", columnvalue.getv1()));
    rowPutChange.addColumn(new Column("v2", columnvalue.getv2()));
    client.putRow(new PutRowRequest(rowPutChange));
}
//DeleteRow
private static void deleteRow(SyncClient client, String pkValue) {
    PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
    primaryKeyBuilder.addPrimaryKeyColumn(PRIMARY_KEY_NAME, PrimaryKeyValue.fromString(pkValue));
    PrimaryKey primaryKey = primaryKeyBuilder.build();
    RowDeleteChange rowDeleteChange = new RowDeleteChange(TABLE_NAME, primaryKey);
    client.deleteRow(new DeleteRowRequest(rowDeleteChange));
}

注意:涉及主键的Update,需要查分成Delete+Put两步操作
_010
->
_011

源码参考

下载

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:

阿里云存储基于飞天盘古2.0分布式存储系统,产品多种多样,充分满足用户数据存储和迁移上云需求。

官方博客
链接