数据同步-从MySQL到Tablestore

系统管理员 2022-04-24 14:16 255阅读 0赞

# 数据同步-从MySQL到Tablestore #

DataX是阿里集团广泛使用的离线数据导出工具， 本文将详细介绍如何从MySQL导出全量数据到Tablestore（OTS）中。

# 一、导出步骤 #

DataX工具目前已经在github上开源，可以从github上拉到源代码进行本地编译，也可以直接下载编译好的压缩包进行解压直接使用，这里选择本地编译方式。

## 1.下载源代码或压缩包 ##

本机装好git工具后，直接执行以下操作：

git clone https://github.com/alibaba/DataX.git

如果要选择下载压缩包的方式，可以从DataX的github地址上获得下载链接：

https://github.com/alibaba/DataX

或者直接下载： [DataX下载地址][DataX]  
下载完压缩包之后请直接解压缩，并直接进入步骤3。

## 2. Maven打包 ##

cd到下载的源码目录，然后执行：

mvn -U clean package assembly:assembly -Dmaven.test.skip=true

编译完成后， 在/target/datax/datax目录下会观察到如下几个目录：  
bin    conf     job     lib    log    log\_perf    plugin    script    tmp  
  
  
bin目录中存放着可执行的datax.py文件，是整个DataX工具的入口  
plugin目录中是支持各种类型数据源的reader和writer  
conf中主要是存放core.json文件，文件中定义了一些缺省参数值如channle流控、buffer大小等参数，一般不随意修改。  
  
  
注意：此步骤会在本地编译各种数据源的writer和reader，会花费比较长的时间，需要耐心等待。

## 3. 准备全量导出的json文件 ##

{
        "job": {
            
            "content": [
                {
                    "reader": {
                        "name": "mysqlreader", #指定使用mysqlreader读取
                        "parameter": {
                            "username": "username",#mysql用户名
                            "password": "password",#mysql密码
                            "connection": [
                                {
                                    "querySql": [ #指定执行的SQL语句
                                        "select bucket_name, delta , timestamp ,cdn_in, cdn_out ,total_request from vip_quota where bucket_name='xxx' "
                                    ],
                                    "jdbcUrl": ["jdbc:mysql://10.10.0.8:3306/db1?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true" #jdbc连接串
                                    ]
                                }
                            ]
                        }
                    },
    
                    "writer": {
                        "name": "otswriter",#指定使用otswriter进行写入
                        "parameter": {#数据源配置
                            "endpoint":"https://smoke-test.xxxx.ots.aliyuncs.com",#ots实例的endpoint
                            "accessId":"xxxx",
                            "accessKey":"xxxx",
                            "instanceName":"smoke-test",#实例名
                            "table":"vip_quota",#写入目标的table名称
                            "primaryKey":[#主键名称和类型
                                {"name":"bucket_name", "type":"string"},
                                {"name":"delta", "type":"int"}
                                {"name":"timestamp", "type":"int"}
                            ],
                            "column":[#其它column的名称和类型
                                {"name":"cdn_in","type":"int"},
                                {"name":"cdn_out","type":"int"},
                                {"name":"total_request","type":"int"},
                            ],
                            "writeMode":"UpdateRow"#写入模式
                        }
                    }
    
                }
            ]
        }
    }

以上为querySql模式导出。  
或者，也可以配置成如下的table模式导出：

{
        "job": {
            "setting": {
                "speed": {
                     "channel": 3  #指定channel个数，这个参数跟并发数密切相关
                },
                "errorLimit": {#容错限制
                    "record": 0,
                    "percentage": 0.02
                }
            },
            "content": [
                {
                    "reader": {
                        "name": "mysqlreader",#指定使用mysqlreader读取
                        "parameter": {
                            "username": "username",#mysql用户名
                            "password": "password",#mysql密码
                            "column": [  #table模式下可以指定需要查询哪些列
                                "bucket_name",      
                                "timestamp" ,
                                "delta" , 
                                "cdn_in", 
                                "cdn_out" ,
                                "total_request"
                            ],
                            "splitPk": "timestamp",#指定split字段
                            "connection": [
                                {
                                    
                                    "table": [#导出的表名
                                        "vip_quota"
                                     ],
                                    "jdbcUrl": ["jdbc:mysql://10.10.1.7:3306/db1"#jdbc连接串
                                    ]
                                }
                            ]
                        }
                    },
    
                    "writer": {
                        "name": "otswriter",#指定使用otswriter进行写入
                        "parameter": {#数据源配置
                            "endpoint":"https://smoke-test.xxxx.ots.aliyuncs.com",#ots实例的endpoint
                            "accessId":"xxx",
                            "accessKey":"xxx",
                            "instanceName":"smoke-test",#实例名
                            "table":"vip_quota",#写入目标的table名称
                            "primaryKey":[#主键名称和类型
                                {"name":"bucket_name", "type":"string"},
                                {"name":"delta", "type":"int"}
                                {"name":"timestamp", "type":"int"}
                            ],
                            "column":[#其它column的名称和类型
                                {"name":"haha","type":"int"},
                                {"name":"hahah","type":"int"},
                                {"name":"kengdie","type":"int"},
                            ],
                            "writeMode":"UpdateRow"#写入模式
                        }
                    }
    
                }
            ]
        }
    }

上述配置文件中，可以看到，该json文件定义了一次数据导出导入的数据源信息和少量系统配置。  
配置主要分两部分：  
setting部分： 主要是speed（跟速率、并发相关）和errorLimit（容错限制）  
content部分：主要是数据源信息， 包含reader和writer两部分  
  
  
同时，配置中的MySQL应该确保执行DataX任务的机器能够正常访问；目标Tablestore表，可以通过控制台或则SDK提前建好。本例中Tablestore的表名为vip\_quota， 定义了由3个column组成的PrimaryKey。

## 4. 执行同步命令 ##

python datax.py  -j"-Xms4g -Xmx4g" mysql_to_ots.json

\-j"-Xms4g -Xmx4g"  可以限制jvm占用内存的大小，如果不指定，将会使用conf/core.json中的配置，默认是1G。

# 二、原理介绍 #

DataX进行数据同步的过程主要包括三部分：

1.  数据源读取
2.  DataX中的数据交换
3.  数据目标端写入

在MySQL导出到Tablestore的场景中，对于MySQL数据源来说，DataX通过MySQL驱动使用reader中的MySQL连接串配置，直接发送SQL语句获取到查询数据，这些数据会缓存在本地jvm中，而后由writer线程将这些数据写入到Tablestore表中。  
  
  
在DataX中，mysqlreader配置有两种模式，一种是table模式，另外一种是querySql模式，两种模式使用起来略有差别。

1.  **table模式**

table模式的json配置文件请观察导出步骤的第3部分。  
table模式下，用户不再需要自己写select语句，而是由DataX根据json中的的column、table、spliPk配置项，自行拼接SQL语句，观察执行日志如下：  
![image1][]  
  
  
  
         在table模式下， channel个数决定了reader和writer的个数上限，假设为m个：  
        如果指定了splitPk字段，DataX会将mysql表中数据按照splitPk切分成n段，n大致为5倍的channel个数（有兴趣的同学可以去阅读一下DataX的源码）。splitPk的字段限制了必需是整型或者字符串类型。由于DataX的实现方式是按照spliPk字段分段查询数据库表，那么spliPk字段的选取应该尽可能的选择分布均匀且有索引的字段，比如主键id、唯一键等字段。DataX会启动m个reader线程，消费DataX切分好的n个查询sql语句（task）， 对应的会有m个writer线程将查询出来的数据写入OTS表中，并行度为m（也就是配置的channel个数）。  
        如果不指定splitPk字段，DataX将不会进行数据的切分，并行度直接退化成1。

需要指出的是，table模式下，如果用户指定了spliPk将数据切分成了n段，由于这些task不是在同一个事务下进行select，那么最终取出的全量数据很有可能是不一致的。为了拿到一致性数据，要么不要配置spliPk使用单线程，要么确保mysql中要导出的数据不会再发生变化。

1.  **querySql模式**

querySql模式一般用于有条件的数据导出。准备步骤中的第一个json文件就是一个典型的querySql模式配置。  
在此模式下，DataX不会再按照指定的column、table参数进行sql的拼接，而是会直接略过这些配置（如果有），直接执行querySql语句，task数量总是1，因此在此模式下channel的配置不再有多线程的效果。

# 三、性能调优 #

有人肯定会有疑问，有什么办法可以尽可能加速数据的导出呢？  
        一般来说，大家首先想到的是提高并发度。在DataX中channel的个数决定了并发数量，但是要使channel参数生效，并不是简单配一下channel就完事了。在MySQL导入Tablestore表的场景下，channel生效仅在能够split出多个SQL语句的场景下，也就是table模式+spliPk下有用。  
  
  
前面提到，DataX的数据同步涉及三部分：  
     1.数据读取  
     2.数据交换  
     3.数据写入  
对于以上三个环节，都有不同的优化方式，分析如下。

## 1.数据读取 ##

对于数据源读取，导出的两种模式：table模式和sqlQuery模式前面做了阐述，这里不再重复。

## 2. 数据交换 ##

对于数据交换，前面提到，发送给MySQL数据库SQL语句后会得到查询的数据集，缓存在DataX的buffer中；除此之外，每个channel也维护了自己的record队列，如果存在并发，channel的个数越多，也会需要更多的内存。因此首先需要考虑的是jvm的内存大小参数，在导出步骤这一节中， -j参数可以用来指定jvm的内存大小。  
除此之外，有几个控制channel的关键参数：  
![image2][]  
  
  
  
以上配置位于conf/core.json中：  
capacity限制了channel中队列的大小（也就是最多缓存record的个数）  
byteCapacity限制了record占用的内存大小，core.json中的默认配置是64MB，若不指定将会被配置为8MB  
这两个参数决定了每个channel能buffer的记录数量和内存占用情况，如果有需要调整，用户应该按照DataX实际的运行环境予以配置。例如MySQL中每个record都比较大，那么可以考虑适当调高byteCapacity，当然调整这个参数还要考虑机器的内存情况。  
  
  
一般情况下，channel队列本身配置的调整并不会很常见，但是对于另外几个流控参数，在使用DataX的时候应该注意。有两个常用的流控参数：  
a. byte  限制通道的默认传输速率， -1表示不限制  
b. record 限制通道的传输记录数，-1表示不限制  
这两个参数都是在flowControlInterval间隔里采样后根据采样值来决定是否流控的。

{
        "core": {  #定义了全局的系统参数，不指定会使用默认值
            "transport": {
                "channel": {
                    "speed": {     
                        "record": 5000,
                        "byte": 102400
                    }
                }
            }
        },
    
    
        "job": {
            "setting": {
                "speed": {  #定义了单个channel的控制参数
                    "record": 10000,
                },
                "errorLimit": {
                    "record": 0,
                    "percentage": 0.02
                }
            },
            "content": [
                {
                    "reader": {
                          .....#省略
                    },
    
                    "writer": {
                        .....#省略
                    }
    
                }
            ]
        }
    }

## 3.数据写入 ##

对于数据写入，Tablestore是基于LSM设计的高性能高吞吐的分布式数据库产品，每一张表，都会被切分成很多的数据分区，分布在不同的服务器上，吞吐能力十分强悍。如果写入能够打散在所有的服务器上面，就能够利用所有服务器的服务能力，更高速地写入，也就是说表分区数量和吞吐能力是正相关的。正常情况下，新建的表默认分区数量都是1，这个数目会随着表的不断写入自动分裂不断增长，但是自动分裂的周期较长，对于新建表马上进行数据导入的情况，单分区很可能不够用导致导入不够顺畅。 推荐的做法，一般是在建表的时候，对表进行预分区，这样可以在一开始导入的时候就获得极好的性能，而不用等自动分裂。  
另外适当的提高批量写入的批次大小（batchWriteCount），也可以有效地提高吞吐率。相关关键配置如下：

{
        "job": {
            "setting": {
               ....#省略
            },
            "content": [
                {
                    "reader": {
                          .....#省略
                    },
    
                    "writer": {
                        "name": "otswriter",
                        "parameter": {
                                .......
                            "writeMode":"UpdateRow",
                            "batchWriteCount":100
                        }
                    }
    
                }
            ]
        }
    }

## 4.总结 ##

综合以上叙述，调优可以从以下几个方面着手：  
1.在可能的情况下，无论是table模式还是sqlQuery模式，选好spliPk，写好where条件， 保证SQL的高效执行  
2.jvm的内存大小要考虑进来，尤其在多channel生效的情况下，内存分配太小会严重限制DataX的吞吐  
3.为了保证安全，可以综合考量channel的个数和流控参数，保证理论峰值不会对服务器产生过高的压力；  
4.为了提升效率，可以适当提高channel的个数从而提高并发数，调高每个channel的byte和record限制，从而提高DataX的吞吐  
5.对目标端Tablestore的表进行预分区，充分利用分布式存储的特点，将写入压力分散到多台机器上，提高写入速度;提高写入batch的大小也可以明显提高吞吐。

# 四、注意事项 #

1.  reader和writer的字段映射关系是通过字段位置一一对应的，而非字段名
2.  writer中的parameter中primaryKey的描述必须和Tablestore中定义的字段名、类型一一匹配，事实上DataX在启动的时候，也会去目标数据源中拉取表定义信息，如果对应不上，会直接抛异常
3.  writer中的parameter中column中字端的数量和类型应该和querySql中select的字段一致，字段名可以不一样；如果没有指定querySql，而是通过table名表示全表导出，writer中的column也应该和table表中的字段对应

[Github参考][Github]

[DataX]: http://datax-opensource.oss-cn-hangzhou.aliyuncs.com/datax.tar.gz
[image1]: /images/20220214/09b7d41f0ff0457f85ad1082e3a18f08.png
[image2]: /images/20220214/338a976ad9db41dda21798a0191c48c3.png
[Github]: https://github.com/aliyun/tablestore-examples/tree/master/tools/Datax-MySQL2TableStore