使用DataX进行OTS实例间数据迁移-阿里云开发者社区

开发者社区> 小q小小倩> 正文

使用DataX进行OTS实例间数据迁移

简介: 本文主要说明在业务不停服的前提下如何实现对OTS的增量迁移,以及迁移后的数据一致性校验
+关注继续查看

 表格存储是 NoSQL 的数据存储服务,是基于云计算技术构建的一个分布式结构化和半结构化数据的存储和管理服务。表格存储的数据模型以二维表为中心。表有行和列的概念,但是与传统数据库不一样,表格存储的表是稀疏的,每一行可以有不同的列,可以动态增加或者减少属性列,建表时不需要为表的属性列定义严格的 schema。

1. 概述

 OTS的数据迁移可以使用DataX完成全量数据迁移。但由于部分数据表的数据量较大,无法在指定的窗口内完成全量迁移,且目前DataX只能针对主键值进行范围查询,暂不支持按照属性列范围抽取数据。所以可以按如下两种方式实现全量+增量的数据迁移:

  1. 分区键包含范围信息(如时间信息、自增ID),则以指定range为切分点,分批次迁移;
  2. 分区键不包含范围信息,则可以采用在应用侧双写的模式将数据分批次迁移,写入目标环境同一张业务表。利用OTS的主键唯一性,选择对重复数据执行覆盖原有行的策略来保证数据唯一性;
    本文以应用侧调整为双写模式为例,详细说明OTS数据迁移、校验过程。其中OTS数据迁移流程具体如下所示:

108

1) 预迁移阶段:双写模式中的大表全量迁移。
2) 正式迁移阶段:双写模式中的增量表全量迁移、其余小表的全量迁移。

2. 预迁移阶段

2.1 准备工作

 为保证新老环境的数据一致性,需要在开始数据迁移前,对目标环境的OTS数据表进行数据清空操作,Delete操作是通过Datax工具直接删除表内数据,无需重新建表。具体操作如下:

2.1.1 配置datax任务

 在使用datax执行数据清空前,需配置对应数据表使用datax执行delete任务所需的json文件。在清空数据的配置中,reader与writer均配置目标端的连接信息,且数据写入模式配置DeleteRow即可,具体内容如下。

{
    "job": {
        "setting": {
            "speed": {
                "channel": "5"
            }
        },
        "content": [{
            "reader": {
                "name": "otsreader",
                "parameter": {
                    "endpoint": "http://xxx.vpc.ots.yyy.com/",
                    "accessId": "dest_accessId",
                    "accessKey": "dest_accessKey",
                    "instanceName": " dest_instanceName",
                    "table": " tablename ",
                    "column": [{
                            "name": "xxxxx"
                        },
                        {
                            "name": "xxxxx"
                        }
                    ],
                    "range": {
                        "begin": [{
                            "type": "INF_MIN"
                        }],
                        "end": [{
                            "type": "INF_MAX"
                        }]
                    }
                }
            },
            "writer": {
                "name": "otswriter",
                "parameter": {
                    "endpoint": "http://xxx.vpc.ots.yun.yyy.com/",
                    "accessId": "dest_accessId",
                    "accessKey": "dest_accessKey",
                    "instanceName": " dest_instanceName",
                    "table": " tablename ",
                    "primaryKey": [{
                        "name": "xxxxx",
                        "type": "string"
                    }],
                    "column": [{
                            "name": "xxxxx",
                            "type": "string"
                        },
                        {
                            "name": "xxxxx",
                            "type": "string"
                        }
                    ],
                    "writeMode": "DeleteRow"
                }
            }
        }]
    }
}

2.1.2 执行datax任务

  1. 登录datax所在ECS后,进入datax所在路径
  2. 在对应的工具机分别执行del_pre.sh脚本,即可开始目标环境对应表的数据清空,具体命令如下:
sh del_pre.sh

del_pre.sh脚本内容如下:

#!/bin/bash
nohup python datax.py del_table_1.json --jvm="-Xms16G -Xmx16G" > del_table_1.log &

2.2 数据迁移

 在不停服务的情况下把源环境内数据量较大的数据表全部迁移到目标环境内对应的数据表。

2.2.1 配置datax任务

 在datax对数据表配置相应的json文件,迁移配置的具体内容如下:
需注意,由于OTS本身是NoSQL系统,在迁移数据的配置中,必须配置所有的属性列,否则会缺失对应属性列的值。

{
    "job": {
        "setting": {
            "speed": {
                "channel": "5"
            }
        },
        "content": [{
            "reader": {
                "name": "otsreader",
                "parameter": {
                    "endpoint": "http://xxx.vpc.ots.yyy.com/",
                    "accessId": "src_accessId",
                    "accessKey": "src_ accessKey ",
                    "instanceName": " src_instanceName",
                    "table": "tablename",
                    "column": [{
                            "name": "xxxxx"
                        },
                        {
                            "name": "xxxxx"
                        }
                    ],
                    "range": {
                        "begin": [{
                            "type": "INF_MIN"
                        }],
                        "end": [{
                            "type": "INF_MAX"
                        }]
                    }
                }
            },
            "writer": {
                "name": "otswriter",
                "parameter": {
                    "endpoint": "http://xxx.vpc.ots.yun.zzz.com/",
                    "accessId": "dest_accessId",
                    "accessKey": "dest_accessKey",
                    "instanceName": " dest_instanceName",
                    "table": " tablename ",
                    "primaryKey": [{
                        "name": "xxxxx",
                        "type": "string"
                    }],
                    "column": [{
                            "name": "xxxxx",
                            "type": "string"
                        },
                        {
                            "name": "xxxxx",
                            "type": "string"
                        }
                    ],
                    "writeMode": "PutRow"
                }
            }
        }]
    }
}

2.2.2 执行datax任务

  1. 登录datax所在ECS后,进入datax所在路径
  2. 在对应的工具机分别执行pre_transfer.sh脚本,即可开始专有域OTS到专有云OTS的数据迁移,具体命令如下:
sh pre_transfer.sh
  • pre_transfer.sh脚本内容如下:
#!/bin/bash
nohup python datax.py table_1.json --jvm="-Xms16G -Xmx16G" >table_1.log &

3. 正式迁移阶段

3.1 OTS数据静默

 OTS的数据静默主要是通过观察对应表的数据是否存在变化来判断,校验方式主要包括行数统计、内容统计。

3.1.1 行数统计

 因OTS本身不提供count接口,所以采用在hive创建OTS外部表的方式,读取OTS数据并计算对应数据表的行数,具体操作如下:

  1. 创建外部表
     启动hive,创建上述数据表对应的外部表;为提高统计效率,外部表可以只读取OTS的主键列,建表语句示例如下:
CREATE EXTERNAL TABLE t_oilcard_expenses_old
(h_card_no string)
STORED BY 'com.aliyun.openservices.tablestore.hive.TableStoreStorageHandler'
WITH SERDEPROPERTIES(
"tablestore.columns.mapping"="card_no")
TBLPROPERTIES ("tablestore.endpoint"="$endpoint ","tablestore.instance"="instanceName","tablestore.access_key_id"="ak","tablestore.access_key_secret"="sk","tablestore.table.name"="tableName");
  1. 进入脚本所在路径
    登录Hadoop集群master所在ECS,进入hive所在目录;
  2. 执行行数统计
    执行pre_all_count.sh脚本,即可开始源环境内OTS对应表的行数统计;
nohup sh pre_all_count.sh >pre_all_count.log &
  • pre_all_count.sh脚本内容如下:
#!/bin/bash
  ./bin/hive -e "use ots;select count(h_card_no) from tableName;" >table.rs &

 连续执行两次行数统计,若两次统计结果一致则说明数据已经静默,数据写入以停止;

3.1.2 内容统计

由于部分数据表分区键对应的值比较单一,导致数据全部存储在同一个分区。若采用hive统计行数会耗时太久,所以对于这个表使用datax将OTS数据导入oss的方式进行内容统计,具体操作如下:

  1. 进入脚本所在路径
    登录上述表格对应的ECS,进入datax所在路径;
  2. 执行内容校验
    1) 执行check_table.sh脚本,即可将源环境内OTS数据表导出到OSS;
sh check_table.sh
  • check_table.sh脚本内容如下:
#!/bin/bash
nohup python datax.py table.json --jvm="-Xms8G -Xmx8G">ots2oss01.log &

2) 获取OSS object的ETAG值,写入对应文件table_check01.rs
连续执行两次内容统计,对比两次导出object的ETAG值,若结果一致则说明数据已经静默,数据写入以停止;

3.2 OTS数据迁移

3.2.1 准备工作

 为保证迁移后新老环境数据一致,防止目标环境因测试产生遗留脏数据,在进行数据迁移前,需要将目标环境的OTS的其余全量表进行数据清空。数据清空方式主要有drop、delete,两者的区别如下:

操作方式 优势 劣势 建议
Drop 耗时短 需重新建表
新表分区数需重新扩展
分区数为1的数据表建议选择drop
Delete 分区数保留 耗时较长 需要保留分区数提升迁移速率的表建议选择delete
3.2.1.1 Drop表操作
  1. 登录OTS图形化客户端所在工具机,使用如下信息连接指定OTS实例,并进行对应表的drop操作;
AK: dest_accessId
SK: dest_accessKey
InstanceName: InstanceName
Endpoint:endpoint
  1. 确认删除后,再在客户端重新创建对应的数据;

3.2.1.2 Delete表操作

 Delete操作是通过Datax工具直接删除表内数据,无需重新建表。Datax所需的配置文件参考2.1.1所示。

  1. 登录datax所在ECS后,进入datax所在路径
  2. 在对应的工具机分别执行delete脚本,即可开始目标环境OTS的对应表的数据清空,具体命令如下:
sh del_table_01.sh
  • del_table_01.sh脚本内容如下:
#!/bin/bash
nohup python datax.py del_table.json --jvm="-Xms16G -Xmx16G">del_table.log &

3.2.2 数据迁移

 在源环境停止服务的情况下把双写模式中的增量表全量迁移以及其余小表全部迁移到目标环境内对应的数据表。
具体操作如下:

3.2.2.1 配置datax任务

 在datax对上述数据表配置相应的json文件,迁移配置的具体内容参考2.2.1,在迁移数据的配置中,需要列全所有的属性列。

3.2.2.2 执行datax任务
  1. 登录datax所在ECS后,进入datax所在路径
  2. 在对应的工具机分别执行transfer.sh脚本,即可开始专有域OTS到专有云OTS的数据迁移,具体命令如下:
sh transfer.sh
  • transfer.sh脚本内容如下:
#!/bin/bash
nohup python datax.py Table.json  >Table.log &

3.3 OTS数据校验

 新老环境OTS的数据校验方式均包括行数统计、内容统计,具体如下。

3.3.1 源环境数据统计

 源环境OTS数据表的数据量统计依据数据静默期间最后一次的统计结果即可。

3.3.2 目标环境数据统计

3.3.2.1 行数统计

 因OTS本身不提供count接口,且目标环境ODPS支持创建OTS外部表,所以采用在ODPS创建OTS外部表的方式,读取OTS数据并计算对应数据表的行数,具体操作如下:

  1. 创建外部表
    登录odpscmd,创建上述数据表对应的外部表;
  2. 进入脚本所在路径
    登录odpscmd工具所在ECS,进入odps所在路径;
  3. 执行行数统计
    执行newots_count.sh脚本,即可进行目标环境内OTS对应表的行数统计;
nohup sh newots_count.sh >newots_count.log &
  • newots_count.sh脚本内容如下:
#!/bin/bash
./bin/odpscmd -e "select count(h_card_no) from tableName;" >table.rs &
3.3.2.2 内容统计

由于源环境的部分数据表采用内容统计的方式进行数据校验,为了方便对比数据是否一致,所以目标环境也采用内容统计的方式,具体操作参考3.1.2。

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

相关文章
centos安装gitlab并进行数据迁移
centos安装gitlab并进行数据迁移
2757 0
MySQL超时参数以及相关数据集成、DataX数据同步案例分享
一、背景 MySQL系统变量提供关于服务器的一些配置和能力信息,大部分变量可在mysqld服务进程启动时设置,部分变量可在mysqld服务进程运行时设置。合理的系统变量设值范围,是保障MySQL稳定提供服务的重要因素。
12905 0
在医疗保健中使用AI:将数据转化为行动
在医疗保健领域,不乏支持人工智能(AI)好处的统计、调查和行业炒作。人工智能已经在我们的日常生活中发挥了重要作用。医疗保健可能是人工智能对我们日常生活影响最大的领域之一。尽管如此,我们才刚刚开始了解人工智能在医疗服务方面的作用。
215 0
【Spring Boot】使用JDBC 获取相关的数据
【Spring Boot】使用JDBC 获取相关的数据 使用JDBC 获取相关的数据 什么是JDBC Java Database Connectivity 是一种用于执行SQL语句的Java API,与数据库建立连接、发送 操作数据库的语句并处理结果。
4726 0
谈谈Spring Boot 数据源加载及其多数据源简单实现
业务需求 提供所有微服务数据源的图形化维护功能 代码生成可以根据选择的数据源加载表等源信息 数据源管理要支持动态配置,实时生效附录效果图 实现思路 本文提供方法仅供类似简单业务场景,在生产环境和复杂的业务场景 请使用分库分表的中间件(例如mycat)或者框架 sharding-sphere (一直在用)等 先来看Spring 默认的数据源注入策略,如下代码默认的事务管理器在初始化时回去加载数据源实现。
1064 0
使用EMR-Kafka Connect进行数据迁移
流式处理中经常会遇到Kafka与其他系统进行数据同步或者Kafka集群间数据迁移的情景。使用EMR Kafka Connect可以方便快速的实现数据同步或者数据迁移。本文介绍使用EMR Kafka Connect的REST API接口在Kafka集群间进行数据迁移。
13224 0
运维编排场景系列-----给ECS实例自动打TAG
Tag简介和场景 对于一般的资源管理需求,都是针对一个用户下数量较多的情况,当实例等数量较多时,对实例进行运维管理等操作就会变得比较困难,有时候甚至需要采取拆分账号的方式管理不同部门或者不同用途的资源。
1523 0
穿越数据的变迁 - 如何使用阿里云Redis数据闪回进行按时间点的数据恢复
如何使用阿里云Redis数据闪回进行按时间点的数据恢复
1728 0
7
文章
1
问答
文章排行榜
最热
最新
相关电子书
更多
《SaaS模式云原生数据仓库应用场景实践》
立即下载
文娱运维技术
立即下载
《看见新力量:二》电子书
立即下载