ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 数据库 > MySQL > mysql亿级大表重构方案介绍

mysql亿级大表重构方案介绍

原创 MySQL 作者：caoyutingtjpu 时间：2017-05-11 16:40:53 0 删除 编辑

mysql亿级大表重构方案介绍

 作者：sylar版权所有[文章允许转载,但必须以链接方式注明源地址,否则追究法律责任.]

【情况简述】

   本文主要分享的博主将mysql生产环境上亿大表按照一定规则拆分成若干个小表并迁移的思路、实现方式、注意事项等等。

【背景说明】

   生产环境favourite表5.8亿，情况如下：

表名

表结构

rows

数据库版本

favourite

CREATE TABLE `favourite` (

  `id` int(11) NOT NULL AUTO_INCREMENT,

`user_id` int(11) NOT NULL,

`target_type` int(11) NOT NULL,

`target_id` int(11) NOT NULL,

`created_at` datetime NOT NULL,

`status` smallint(6) NOT NULL DEFAULT '0',

PRIMARY KEY (`id`),

  UNIQUE KEY `uniq_user_target` (`user_id`,`target_type`,`target_id`),

  KEY `idx_targetid` (`target_id`)

) ENGINE=InnoDB  DEFAULT CHARSET=utf8mb4

587312519

5.7.12


  下面sql因表的量级变的比较慢，已无法通过调整索引或调整sql进行优化：

SQL

time

SELECT count(1) AS count_1 FROM `favourite` WHERE `favourite`.target_id = 636 AND `favourite`.target_type = 1

4.7S

SELECT `favourite`.target_id AS `favourite_target_id` FROM `favourite` WHERE `favourite`.user_id = 338072 AND `favourite`.target_type = 0 AND `favourite`.status = 0 ORDER BY `favourite`.id DESC

2.25S

DELETE FROM favourite WHERE user_id = 17327373 AND target_id = 917 AND target_type = 1

0.9S

  为了业务响应比较快，决定拆分favourite表。经过业务沟通，user_id使用较为频繁，故通过user_id拆分，拆分规则根据user_id%1024打算到1024表，映射关系如下：

user_id%1024 =0  =>favourite_0000

user_id%1024 =1  =>favourite_0001

user_id%1024 =2  =>favourite_0002

……

user_id%1024 =1023  =>favourite_1023


注意：

1)拆分一定要根据业务情况来决定，不能一概而论！


【思路说明】

1、配置好canal ，canal是阿里开源的获取binlog信息的软件。从第一步开始到最后结束，canal一直不停获取binlog信息。

2、在不影响业务的数据库上（此处用的从库）将favourite导出成1024个表对应的文件

3、将导出备份文件导入生产环境

4、将canal获取的数据导入到1024个分表（一直进行直到结束）

5、待分表数据与原大表数据差不多时，在业务不繁忙时，切favourite业务读操作

6、切生产favourite写操作

7、待canal无新的记录产生，整个业务切换完毕

8、结束


注意：

1)使用canal获取binlog信息，注意参数设置为

    binlog_format=row

    binlog_row_image=full（默认是FULL，以防有些实例设置为minimal）

     binlog_row_image=minimal，此时主库进行delete from a where target_id =，在binlog记录的也是这样的操作，而如果target_id是唯一索引，则到了binlog记录的是delete from a where id（id是主键）。同样生产是delete favourite 是根据`user_id`,`target_type`,`target_id`，根据前面所说记录到binlog的是 delete from favourite where id，canal需要在业务发生切换时候记录变更的情况，可是获取的是删除的id，新的分表的id不会跟旧的favouriteid相同（因为旧的favourite的id已经到了8亿多，新的分表沿用旧表id是很浪费的情况）并不知道user_id是哪个，无法对新表进行同步。

2)服务先切读再切写的这个方式，在切换时间内会有一定数据误差的。比如用户在业务切换读操作的同时，往favourite表插入了一条数据；可是写操作没有切过来，数据落入旧的favourite表，用户就会发现异常影响用户的体验。这个情况直到canal将所有信息同步到新的分表后才解决。如果自己的业务无法接受这个情况，可以根据实际情况调整，不一定按照此方式切换服务。


【方案说明】

   根据上面思路，需要解决两个问题

1）如何配置canal。canal是阿里推出的获取binlog的开源产品，我们此次canal调用是java工程师帮忙写的代码，博主不懂java，故省略canal配置信息。

2）如何将favourite导出成1024个分表所需的数据，然后倒入指定库。

下面主要说明问题2实现的方式，一共有两种：

items

方案一

方案二

实现手段

mysqldump

mycat

拆分耗时

4.5Hour

2Hour

准备时间

3Hour,需要加函数索引

<1Hour,准备mycat环境和mycat对应的数据库

优点

不需要配置mycat环境

时间比方案一节省2Hour,导入目标环境后不需在初始化id

缺点

耗时太久、导入目标环境后还需要初始化id

需要熟悉mycat配置、分库规则

具体方式

Step1.在从库建立函数索引,耗时3Hour

Step2.在从库使用mysqldump的--where参数导出

Step3.导入目标库并初始化id

Step1.搭建mycat环境，并配置好相关规则

Step2.使用mysqldump备份文件

Step3.将备份文件导入mycat

Step4.在mycat对应库初始化id

Step5.将处理后的文件导入目标库

【方式一:mysqldump】

Step1.在从库建立函数索引,耗时3Hour

   alter table favourite  add `vis_user_id` int(11) GENERATED ALWAYS AS ((`user_id` % 1024)) STORED;

注意：

1）要在从库建立函数索引，影响会降低很多，如果能把让生产不访问该从库更好。确保生产环境访问该从库时没有select * from favourite where …..这样的命令

2）如果数据库版本低于5.7无法使用函数索引，那么step2.mysqldump备份一次开启4个并发进程，一次耗时230秒；如果有索引，则为30-60秒

 

Step2.在从库使用mysqldump的--where参数导出

思路：

1）使用--where=" user%1024=0001"导出成按拆分规则命名的文件，该例子对应文件名为0001.sql，一共会产生1024个这样的文件。
2）然后根据导出的文件名用sed命令替换表名（sed是shell命令）

具体脚本如下：


点击(此处)折叠或打开

  1. #查看log日志调用并发函数实现多线程备份
  2. function dumpAllTable()
  3. {
  4.         local schemaFile="${parallel_log}"
  5.         #最大的表先备份（因多进程并发，最短完成时间依赖于最大表的完成）
  6.         allTable=`cat $schemaFile | wc -l`
  7.         i_import=0
  8.         declare -a array_cmds
  9.         i_array=0
  10.         while read file; do
  11.                 i_import=`expr $i + 1`
  12.                 array_cmds[i_array]="${file}"
  13.                 i_array=`expr ${i_array} + 1`
  14.         done < $schemaFile
  15.         execConcurrency "${threadsNum}" "${array_cmds[@]}"
  16. }


  19. #并发函数
  20. function execConcurrency()
  21. {
  22.         #并发数据量
  23.         local thread=$1
  24.         #并发命令
  25.         local cmd=$2
  26.         #定义管道，用于控制并发线程
  27.         tmp_fifofile="/tmp/$$.fifo"
  28.         mkfifo $tmp_fifofile
  29.         #输入输出重定向到文件描述符6
  30.         exec 6<>$tmp_fifofile
  31.         rm -f $tmp_fifofile
  32.         #向管道压入指定数据的空格
  33.         for ((i=0;i<$thread;i++)); do
  34.                 echo
  35.         done >&6
  36.         #遍历命令列表
  37.         while [ "$cmd" ]; do
  38.                 #从管道取出一个空格（如无空格则阻塞，达到控制并发的目的）
  39.                 read -u6
  40.                 #命令执行完后压回一个空格
  41.                 { eval $2;echo >&6; } & #> /dev/null 2>&1 &
  42.                 shift
  43.                 cmd=$2
  44.         done
  45.         #等待所有的后台子进程结束
  46.         wait
  47.         #关闭df6
  48.         exec 6>&-
  49. }

  51. function instance_log()
  52. {
  53.          if [[ -f "${parallel_log}" ]];then
  54.                 rm -rf "${parallel_log}"
  55.          fi;
  56.          for i in $(seq -w 0 1023)
  57.             do
  58.             echo "${MYSQLDUMP} -u${user} -p${password} -S ${socket} --default-character-set=utf8mb4 -c --set-gtid-purged=OFF --skip-add-locks --skip-quick --no-create-db --log-error=${backupdir}/cyt.log --skip-add-drop-table ${database} ${tablename} --where=\" vis_${columnname}=${i} \"> ${backupdir}/${i}.sql " >>${parallel_log}
  59.          done
  60.          dumpAllTable
  61.          mv ${parallel_log} ${parallel_log}_bak
  62. }

  64. function change()
  65. {
  66.          if [[ -f "${parallel_log}" ]];then
  67.                 rm -rf "${parallel_log}"
  68.          fi;
  69.          for i in $(seq -w 0 1023)
  70.              do
  71.              echo "sed -i 's/\`${tablename}\`/\`${tablename}_${i}\`/g' ${backupdir}/${i}.sql" >>${parallel_log}
  72.          done
  73.          dumpAllTable
  74.          mv ${parallel_log} ${parallel_log}_bak_2
  75. }

  77. function main()
  78. {
  79.          MYSQLDUMP=""
  80.          user=""
  81.          password=""
  82.          socket=""
  83.          columnname="user_id"
  84.          tablename="favourite"
  85.          database="kuaikan"
  86.          backupdir="/data/backup"
  87.          parallel_log=""
  88.          threadsNum=2
  89.          instance_log
  90.          change
  91. }

  93. main

      在加了函数索引情况下，开启4个并行进行dump，一次花费60秒左右，一共1024个文件：1024／4*60s/3600s=4.3hour 时间长达4个半小时。（根据自己机器IO来调整并行度，不是并行越高，速度越快，博主环境在开启6个并行后，预计耗时5个多小时。开启2个并行耗时跟4个并行差不多）

Step3.导入目标库并初始化id

    注意新表的id都比较大，需要重新初始化下id,这个表的created_at跟id顺序是一样的，故按照created_at排序，初始化方式为：

i)alter table favourite_XXXX drop column id ,drop column vis_user_id,drop key uniq_user_target ,order by created_at ;

#因为需要初始化id，所以需要先删除id，同时删除唯一索引是为了后面语句order by created_at。因为InnoDB会对主键建立聚簇索引，如果没有主键，InnoDB会用一个具有唯一且非空值的索引来代替，此时就是UNIQUE KEY `uniq_user_target`。一般来说，DBMS都会以聚簇索引的形式来存储实际的数据，它是其它二级索引的基础。在聚簇索引中，索引树的叶级页包含实际的数据：记录的索引顺序与物理顺序相同。在非聚簇索引中，叶级页指向表中的记录：记录的物理顺序与逻辑顺序没有必然的联系。所以为了让表按照created_at排序，需要删除上面的唯一键，否则执行下面语句无法生效。

ii)alter table favourite_XXXX  add id  int(11) NOT NULL AUTO_INCREMENT first ,add primary key (id), add UNIQUE KEY uniq_user_target(user_id,target_type,target_id)

#初始化id，补回索引。

iii)具体脚本如下


点击(此处)折叠或打开

  1. function import_log()
  2. {
  3.          if [[ -e "${log_1}" ]];then
  4.             rm -rf ${log_1};
  5.          fi;
  6.          for filename in `du -sk ${BACKUP_DIR}/*.sql | sort -nr |awk '{print $2}'`
  7.             do
  8.             
  9.             echo "mysql -u${DB_USER} -p${DB_PASSWORD} --port=${port} --socket=${socket} --host=${host} -A ${database} --tee=/data/pat0405.log -e \"source ${filename}\"">>${log_1};
  10.           done
  11.           import=`echo $?`
  12.           if [[ "${import}" -eq 0 ]];then
  13.              echo "import_log successfully" `date "+%Y-%m-%d %H:%M:%S"`>>${BACKUP_DIR}/import.log 2>&1
  14.           else
  15.              echo "import_log failed" `date "+%Y-%m-%d %H:%M:%S"`>>${BACKUP_DIR}/import.log 2>&1
  16.           fi
  17. }

  19. function modify_column_log()
  20. {
  21.          if [[ -e "${log_2}" ]];then
  22.             rm -rf ${log_2};
  23.          fi;
  24.          for i in $(seq -w 0 1023 )
  25.              do
  26.              echo "mysql -u${DB_USER} -p${DB_PASSWORD} --port=${port} --socket=${socket} --host=${host} -A ${database} --tee=/data/pat0405.log -e \" alter table favourite_${i} drop column id ,drop key uniq_user_target ,order by created_at ;alter table favourite_${i} add id int(11) NOT NULL AUTO_INCREMENT first ,add primary key (id), add UNIQUE KEY uniq_user_target(user_id,target_type,target_id)\" " >>${log_2}
  27.          done
  28.          modify=`echo $?`
  29.          if [[ "${modify}" -eq 0 ]];then
  30.              echo "modify_column_log successfully" `date "+%Y-%m-%d %H:%M:%S"`>>${BACKUP_DIR}/import.log 2>&1
  31.          else
  32.              echo "modify_column_log failed" `date "+%Y-%m-%d %H:%M:%S"`>>${BACKUP_DIR}/import.log 2>&1
  33.          fi
  34. }



  38. #调用函数log，查看log日志调用并发函数实现多线程备份
  39. function dumpAllTable()
  40. {
  41.         local schemaFile="$1"
  42.         local threadsNum="$2"
  43.         #最大的表先备份（因多进程并发，最短完成时间依赖于最大表的完成）
  44.         allTable=`cat $schemaFile | wc -l`
  45.         i_import=0
  46.         declare -a array_cmds
  47.         i_array=0
  48.         while read file; do
  49.                 i_import=`expr $i + 1`
  50.                 array_cmds[i_array]="${file}"
  51.                 i_array=`expr ${i_array} + 1`
  52.         done < ${schemaFile}
  53.         execConcurrency "${threadsNum}" "${array_cmds[@]}"
  54.         delta_ret_val=`echo $?`
  55.         if [[ "${delta_ret_val}" -eq 0 ]];then
  56.            echo "${schemaFile} successfully" `date "+%Y-%m-%d %H:%M:%S"`>>${BACKUP_DIR}/import.log 2>&1
  57.         else
  58.            echo "${schemaFile} failed" `date "+%Y-%m-%d %H:%M:%S"`>>${BACKUP_DIR}/import.log 2>&1
  59.         fi
  60. }


  63. #并发函数
  64. function execConcurrency()
  65. {
  66.         #并发数据量
  67.         local thread=$1
  68.         #并发命令
  69.         local cmd=$2
  70.         #定义管道，用于控制并发线程
  71.         tmp_fifofile="/tmp/$$.fifo"
  72.         mkfifo $tmp_fifofile
  73.         #输入输出重定向到文件描述符6
  74.         exec 6<>$tmp_fifofile
  75.         rm -f $tmp_fifofile
  76.         #向管道压入指定数据的空格
  77.         for ((i=0;i<$thread;i++)); do
  78.                 echo
  79.         done >&6
  80.         #遍历命令列表
  81.         while [ "$cmd" ]; do
  82.                 #从管道取出一个空格（如无空格则阻塞，达到控制并发的目的）
  83.                 read -u6
  84.                 #命令执行完后压回一个空格
  85.                 { eval $2;echo >&6; } & #> /dev/null 2>&1 &
  86.                 shift
  87.                 cmd=$2
  88.         done
  89.         #等待所有的后台子进程结束
  90.         wait
  91.         #关闭df6
  92.         exec 6>&-
  93.         exec_con=`echo $?`
  94.         if [[ "${exec_con}" -eq 0 ]];then
  95.            echo " execConcurrency successfully" `date "+%Y-%m-%d %H:%M:%S"`>>${BACKUP_DIR}/import.log 2>&1
  96.         else
  97.            echo "execConcurrency failed" `date "+%Y-%m-%d %H:%M:%S"`>>${BACKUP_DIR}/import.log 2>&1
  98.         fi

  100. }

  102. function instance()
  103. {
  104.          import_log
  105.          modify_column_log
  106.          echo "-----端口号为"$port"的mysql实例开始按表并发倒入:开始时间为"`date "+%Y-%m-%d %H:%M:%S"`
  107.          #调用执行函数
  108.          dumpAllTable ${log_1} 1
  109.          dumpAllTable ${log_2} 6
  110.          END=`date "+%Y-%m-%d %H:%M:%S"`
  111.          END_T=`date -d "$END" +%s`
  112.          TIME_INVENTAL_M=$[($END_T-$BEGIN_T)/60]
  113.          TIME_INVENTAL_S=$[($END_T-$BEGIN_T)%60]
  114.              

  116. }

  118. #主函数
  119. function main()
  120. {
  121.         #IP地址
  122.         host=""
  123.         port=
  124.         DATE=`date +%F`
  125.         socket=""
  126.         DB_USER=''
  127.         #数据库用户对应的密码
  128.         DB_PASSWORD=''
  129.         #记录开始的时间
  130.         BEGIN=`date "+%Y-%m-%d %H:%M:%S"`
  131.         BEGIN_T=`date -d "$BEGIN" +%s`
  132.         echo '--------------开始按表并发导入:开始时间为 '$BEGIN
  133.         BACKUP_DIR="/data/backup"
  134.         log_1=""
  135.         log_2=""
  136.         database="kk_favourite"
  137.         #调用instance函数
  138.         instance
  139.         succ=`echo $?`
  140.         if [[ "${succ}" -eq 0 ]]&&[[ "${modify}" -eq 0 ]]&&[[ "${import}" -eq 0 ]];then
  141.            echo "MySQL DB Instance:${port} import data sucessfully,cost time:${TIME_INVENTAL_M} min ${TIME_INVENTAL_S} s" >>${BACKUP_DIR}/import.log 2>&1
  142.         else
  143.            echo "MySQL DB Instance:${port} import data failed,cost time:${TIME_INVENTAL_M} min ${TIME_INVENTAL_S} s" >>${BACKUP_DIR}/import.log 2>&1
  144.         fi
  145. }

  147. main "$1" "$2"

        该操作如果导入的实例有其他数据库，一定要在业务不繁忙的时候导入，否则影响生产环境。这是由于dump生成的一条insert相当于一次性插入2-3万行，会占用生产环境的资源。

         此次处理耗时1小时左右（2个并行）

【方案二：mycat分表】

Step1.搭建mycat环境，并配置好相关规则

  mycat分表是一个逻辑库对应一个分表，所以需要创建1024个逻辑库对应1024个分表，考虑到机器IO，打算将1024个库分到4个实例（数据库均是单实例，如果有条件，可以分到更多的机器上，速度会快些）

  需要修改的配置文件：server.xml、schema.xml、rule.xml及其对应的partition-hash-int.txt。具体修改请查看附件conf.zip

conf.zip


注意：

1)需要提前创建1024个逻辑库

2)这里是根据方案一提到的函数索引对应的虚拟列vis_user_id来分的，这样可以直接用mycat的枚举分库，如果不想用虚拟列，可以用mycat hash来划分，这个对于数值划分方式等同于user%1024，这个详情参考mycat权威指南

3)需要提前在四个逻辑库里创建好用于mycat访问的数据库用户

4)在创建完1024个逻辑库后，登入mycat，再创建favourite表，这样每个逻辑库都有该表

 

Step2.使用mysqldump备份文件

   为了能快速导入mycat，故根据mycat分成4个实例规则（可以有误差，不一定要完全一样），导出4份不同数据，以便可以同时4份文件灌入mycat

使用mysqldump导出4个文件，以下备份同时进行，耗时20分钟：

  • mysqldump -u$USERNAME -p$PASSWORD -S $SOCKET --default-character-set=utf8mb4 -c --set-gtid-purged=OFF --skip-add-locks --skip-quick  --no-create-db --log-error=/data/cyt0324.log --skip-add-drop-table    kuaikan favourite --where=" mod(user_id,1024)<256 "     >   /data/favourite_256.sql
  • mysqldump ……………………  --where=" mod(user_id,1024)>=256 and mod(user_id,1024) <512   "     >   /data/favourite_512.sql
  • mysqldump ……………………  --where=" mod(user_id,1024)>=512 and mod(user_id,1024)< 768 "     >   /data/favourite_768.sql
  • mysqldump ……………………  --where=" mod(user_id,1024) >=768"     >   /data/favourite_1024.sql

注意：

1)请在从库或业务不去访问的数据库上进行备份

2)上面设置的参数请根据实际情况调整，一定要加上-c   --skip-add-locks参数，否则导入mycat会异常

Step3.将备份文件导入mycat

   将步骤2导出的四个备份文件同时灌入mycat，整个耗费时间不足90分钟。

注意:

1)请将该操作在后台执行，可以用screen命令来实现。因为mycat是长连接，即使中途断掉后，后面的语句可能还会继续执行，以防出现这样情况，请放到后台执行

 

Step4.在mycat对应库初始化id

此处耗时30分钟

1）在1024分库上初始化id,去掉虚拟列，具体脚本如下


点击(此处)折叠或打开

  1. function dumpAllTable()
  2. {
  3.         local schemaFile="${parallel_log}"
  4.         #最大的表先备份（因多进程并发，最短完成时间依赖于最大表的完成）
  5.         allTable=`cat $schemaFile | wc -l`
  6.         i_import=0
  7.         declare -a array_cmds
  8.         i_array=0
  9.         while read file; do
  10.                 i_import=`expr $i + 1`
  11.                 array_cmds[i_array]="${file}"
  12.                 i_array=`expr ${i_array} + 1`
  13.         done < $schemaFile
  14.         execConcurrency "${threadsNum}" "${array_cmds[@]}"
  15. }


  18. #并发函数
  19. function execConcurrency()
  20. {
  21.         #并发数据量
  22.         local thread=$1
  23.         #并发命令
  24.         local cmd=$2
  25.         #定义管道，用于控制并发线程
  26.         tmp_fifofile="/tmp/$$.fifo"
  27.         mkfifo $tmp_fifofile
  28.         #输入输出重定向到文件描述符6
  29.         exec 6<>$tmp_fifofile
  30.         rm -f $tmp_fifofile
  31.         #向管道压入指定数据的空格
  32.         for ((i=0;i<$thread;i++)); do
  33.                 echo
  34.         done >&6
  35.         #遍历命令列表
  36.         while [ "$cmd" ]; do
  37.                 #从管道取出一个空格（如无空格则阻塞，达到控制并发的目的）
  38.                 read -u6
  39.                 #命令执行完后压回一个空格
  40.                 { eval $2;echo >&6; } & #> /dev/null 2>&1 &
  41.                 shift
  42.                 cmd=$2
  43.         done
  44.         #等待所有的后台子进程结束
  45.         wait
  46.         #关闭df6
  47.         exec 6>&-
  48. }

  50. function instance_log()
  51. {
  52.          if [[ -f "${parallel_log}" ]];then
  53.                 rm -rf "${parallel_log}"
  54.          fi;
  55.          for i in $(seq -w 0 1023 )
  56.              do
  57.              echo "mysql -u$user -p$password -h$IP -P ${port} -A ${database}_${i} -e \" alter table ${tablename} drop column ${columnname}, drop column id ,drop key user_target;alter table ${tablename} order by created_at ;alter table ${tablename} add id int(11) NOT NULL AUTO_INCREMENT first ,add primary key (id); alter table ${tablename} add UNIQUE KEY uniq_user_target(user_id,target_type,target_id)\" " >>${parallel_log}
  58.          done
  59.          dumpAllTable
  60.          mv ${parallel_log} ${parallel_log}_bak
  61. }



  65. function main()
  66. {
  67.          user=""
  68.          password=""
  69.          port=""
  70.          IP=""
  71.          columnname="vis_user_id"
  72.          tablename="favourite"
  73.          database="favourite"
  74.          backupdir="/data/backup"
  75.          parallel_log=""
  76.          threadsNum=6
  77.          instance_log
  78. }

  80. main


2）将1024个分库对应的表并行dump，并使用sed根据库名将表名favourite改为favourite_XXXX，具体脚步如下


点击(此处)折叠或打开

  1. #查看log日志调用并发函数实现多线程备份
  2. function dumpAllTable()
  3. {
  4.         local schemaFile="${parallel_log}"
  5.         #最大的表先备份（因多进程并发，最短完成时间依赖于最大表的完成）
  6.         allTable=`cat $schemaFile | wc -l`
  7.         i_import=0
  8.         declare -a array_cmds
  9.         i_array=0
  10.         while read file; do
  11.                 i_import=`expr $i + 1`
  12.                 array_cmds[i_array]="${file}"
  13.                 i_array=`expr ${i_array} + 1`
  14.         done < $schemaFile
  15.         execConcurrency "${threadsNum}" "${array_cmds[@]}"
  16. }


  19. #并发函数
  20. function execConcurrency()
  21. {
  22.         #并发数据量
  23.         local thread=$1
  24.         #并发命令
  25.         local cmd=$2
  26.         #定义管道，用于控制并发线程
  27.         tmp_fifofile="/tmp/$$.fifo"
  28.         mkfifo $tmp_fifofile
  29.         #输入输出重定向到文件描述符6
  30.         exec 6<>$tmp_fifofile
  31.         rm -f $tmp_fifofile
  32.         #向管道压入指定数据的空格
  33.         for ((i=0;i<$thread;i++)); do
  34.                 echo
  35.         done >&6
  36.         #遍历命令列表
  37.         while [ "$cmd" ]; do
  38.                 #从管道取出一个空格（如无空格则阻塞，达到控制并发的目的）
  39.                 read -u6
  40.                 #命令执行完后压回一个空格
  41.                 { eval $2;echo >&6; } & #> /dev/null 2>&1 &
  42.                 shift
  43.                 cmd=$2
  44.         done
  45.         #等待所有的后台子进程结束
  46.         wait
  47.         #关闭df6
  48.         exec 6>&-
  49. }

  51. function instance_log()
  52. {
  53.          if [[ -f "${parallel_log}" ]];then
  54.                 rm -rf "${parallel_log}"
  55.          fi;
  56.          for i in $(seq -w 0 1023 )
  57.              do
  58.              echo "${MYSQLDUMP} -u${user} -p${password} -h$IP -P $port --default-character-set=utf8mb4 -c --set-gtid-purged=OFF --skip-add-locks --skip-quick --no-create-db --log-error=${backupdir}/cyt.log --skip-add-drop-table ${database}_${i} ${tablename} > ${backupdir}/${i}.sql " >>${parallel_log}
  59.          done
  60.          dumpAllTable
  61.          mv ${parallel_log} ${parallel_log}_bak
  62. }

  64. function change()
  65. {
  66.          if [[ -f "${parallel_log}" ]];then
  67.                 rm -rf "${parallel_log}"
  68.          fi;
  69.          for i in $(seq -w 0 1023 )
  70.              do
  71.              echo "sed -i 's/\`${tablename}\`/\`${tablename}_${i}\`/g' ${backupdir}/${i}.sql" >>${parallel_log}
  72.          done
  73.          dumpAllTable
  74.          mv ${parallel_log} ${parallel_log}_bak_2
  75. }

  77. function main()
  78. {
  79.          MYSQLDUMP="mysqldump"
  80.          user=""
  81.          password=""
  82.          port=""
  83.          IP=""
  84.          columnname="user_id"
  85.          tablename="favourite"
  86.          database="favourite"
  87.          backupdir="/data/backup"
  88.          parallel_log=""
  89.          threadsNum=6
  90.          instance_log
  91.          change
  92. }

  94. main


 Step5.将处理后的文件导入目标库

此处耗时30分钟

将处理后的备份文件导入生产新库，并行导入，并行度最好不要超过3，因为dump导出后insert一次是3万左右，并行度太高，机器IO会hold不住，脚本如下：


点击(此处)折叠或打开

  1. function instance()
  2. {
  3.              log
  4.              echo "-----端口号为"$port"的mysql实例开始按表并发倒入:开始时间为"`date "+%Y-%m-%d %H:%M:%S"`
  5.              #调用执行函数
  6.              dumpAllTable

  8. }



  12. #将要备份的单表从大到小输出到日志里面
  13. function log()
  14. {
  15.             BACKUP_DIR="/data/backup/"
  16.             #过滤掉MySQL自带的DB
  17.             if [ -e ${BACKUP_DIR}/cyt.log ];
  18.                then rm -rf ${BACKUP_DIR}/cyt.log;
  19.             fi;
  20.             for filename in `du -sk ${BACKUP_DIR}/*.sql | sort -nr |awk '{print $2}'`
  21.                do
  22.                a="kk_favourite"
  23.                echo "mysql -u${DB_USER} -p${DB_PASSWORD} --socket=${socket} --host=${host} -A ${a} --tee=/data/pat.log -e \"source ${filename}\"">>$BACKUP_DIR/cyt.log;
  24.              done
  25. }

  27.    

  29. #调用函数log，查看log日志调用并发函数实现多线程备份
  30. function dumpAllTable()
  31. {
  32.         local schemaFile="${BACKUP_DIR}/cyt.log"
  33.         #最大的表先备份（因多进程并发，最短完成时间依赖于最大表的完成）
  34.         allTable=`cat $schemaFile | wc -l`
  35.         i_import=0
  36.         declare -a array_cmds
  37.         i_array=0
  38.         while read file; do
  39.                 i_import=`expr $i + 1`
  40.                 array_cmds[i_array]="${file}"
  41.                 i_array=`expr ${i_array} + 1`
  42.         done < ${BACKUP_DIR}/cyt.log
  43.         execConcurrency "${threadsNum}" "${array_cmds[@]}"
  44.         delta_ret_val=`echo $?`
  45. }





  51. #并发函数
  52. function execConcurrency()
  53. {
  54.         #并发数据量
  55.         local thread=$1
  56.         #并发命令
  57.         local cmd=$2
  58.         #定义管道，用于控制并发线程
  59.         tmp_fifofile="/tmp/$$.fifo"
  60.         mkfifo $tmp_fifofile
  61.         #输入输出重定向到文件描述符6
  62.         exec 6<>$tmp_fifofile
  63.         rm -f $tmp_fifofile
  64.         #向管道压入指定数据的空格
  65.         for ((i=0;i<$thread;i++)); do
  66.                 echo
  67.         done >&6
  68.         #遍历命令列表
  69.         while [ "$cmd" ]; do
  70.                 #从管道取出一个空格（如无空格则阻塞，达到控制并发的目的）
  71.                 read -u6
  72.                 #命令执行完后压回一个空格
  73.                 { eval $2;echo >&6; } & #> /dev/null 2>&1 &
  74.                 shift
  75.                 cmd=$2
  76.         done
  77.         #等待所有的后台子进程结束
  78.         wait
  79.         #关闭df6
  80.         exec 6>&-
  81. }


  84. #主函数
  85. function main()
  86. {
  87.         #获取本地IP地址
  88.         host=""
  89.         port=
  90.         DATE=`date +%F`
  91.         socket=""
  92.         DB_USER=''
  93.         #数据库用户对应的密码
  94.         DB_PASSWORD=''
  95.         #记录开始的时间
  96.         BEGIN=`date "+%Y-%m-%d %H:%M:%S"`
  97.         BEGIN_T=`date -d "$BEGIN" +%s`
  98.         echo '--------------开始按表并发备份:开始时间为 '$BEGIN
  99.         #设置并发备份的线程数
  100.         threadsNum=8
  101.         #调用instance函数
  102.         instance
  103.         echo '--------------backup all database successfully！！！结束时间:' `date "+%Y-%m-%d %H:%M:%S"`
  104. }

  106. main

【小节】

由于方案二速度比方案一节省至少2小时以上的时间，且导入目标库后无需其他处理，故生产环境迁移使用的方案二



来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/31324175/viewspace-2138921/ ，如需转载，请注明出处，否则将追究法律责任。

0
0
上一篇： MongoDB将collection变为shard collection
下一篇： mongoDB Sharded Cluster迁移后更改各个server的IP地址
请登录后发表评论 登录
全部评论
caoyutingtjpu

注册时间：2016-05-13

  • 博文量
    19
  • 访问量
    67026

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593