亿级数据迁移实战方案,真实案例

冷不防 2022-06-08 06:56 209阅读 0赞

背景:  
公司某业务模块的用户通讯记录和通话记录的表，单表记录过亿了,mysql很坚挺好不好！！！！  
其中某表是205773235 约2.0亿  
另外某表是575213155 约5.7亿  
用户数量 约100W  
其中通讯录有的用户是有很多重复记录的,之前设计数据库的同事按条来存的,也就是说一个用户会有多条记录，还有重复的记录(没有去重),历史遗留问题那就不废话了,我们存在的意义就是把不合理的设计去规范化来  
如: id:1 uid:10086 name: 马云爸爸 num:88888  
id:2 uid:10086 name: 马化腾爸爸 num:99999  
这样的数据表设计在数据量少的时候是没有问题的,不去重这点会占用额外的空间，其实这些不太常用而且有很多一对多关系的数据可以存json或者直接用OSS来存的（ps我们技术顾问推荐的存储方式）  
大概思路：  
１.是把100Ｗ的用户拆分成10份，每次循环10Ｗ个用户的数据上传  
２.打通oss接口，文件保存按照日期＋小时这样子保存文件  
３.存储方式json文件方式保存  
4.数据安全方面,用动态密匙加密json，动态的密匙需要按规律保存好（因为文件要开放公共读的,处于对用户数据的安全考虑得加密）  
5.上传成功返回url保存好  
异常处理:  
1.文件上传必然会有上传失败的,每个文件循环上传,如果上传成功,则跳出循环,否则程序休息0.5秒继续上传,最大限度循环10次,如果还是上传失败,把这个用户的uid push到redis的链表里面  
2.文件上传完成后保存上传路径,同理保存也有可能保存失败,也是上面的套路,做一个循环,如果保存一次成功了就跳出循环,否则信息0.5秒继续发起保存sql，成功推出循环,失败10次保存uid到redis上面  
3.每10分钟检查一次redis里面的上传失败或者保存用户数据失败的用户,如果超过500条则继续上传,这里的套路和1.2点同理，如果还是失败,到这里还失败的,其实用户已经失败了20次了,那么应该属于重灾区,把这些用户的uid保存到数据库里面  
4.1.2.3步骤都有和文件日志结合的，我的目标是不要有问题，少有问题！  
最终实践：  
1.其实我没有直接拿用户表的uid,因为有小部分的用户是没有数据的，我创建了一张表,专门来保存符合调教的用户的uid，导入适合的数据的命令如花：

INSERT INTO 某临时表 (uid) 
    select * from (SELECT ct.uid FROM (select max(id) as id from  某符合条件的表啦 group by uid ) pct join  某符合条件的表啦 ct on ct.id=pct.id ) as xxx

2.一步步找到之前的保存数据的方式改成现在的模式  
3.各种测试方法的可行性

**具体迁移方法：**

/**
         *  通话记录迁移 written : yangxingyi date: 2017-09-11
         */
        public function send_phone()
        {
            ini_set('memory_limit','10240M');//限制内存10GB
            $start = trim(I('start'));
            $end   = trim(I('end'));//2017-09-18 15:47 区间1000-100000
            ($start>$end)&&exit("输入范围有误!");
            (!$start||!$end)&&exit("没有输入条件!");
    
            $uids = 某用户表模型->where("id between $start and $end")->field('uid')->select();
            $uids||exit('没有符合条件的数据!');
            file_log('send_phone','正常数据的开始时间','phone');
            $oss = new OssCore();
            $que = 实例化redis;
            $t1 = $t2 = time();//记录耗时
            $bucket = self::$contact_bucket; // 存储空间
    
            $result = $save_res ='';
            $success_count = $false_count = 0;//记录成功失败条数
            $total_count   = count($uids);//记录总条数
            while(list($index,$item)=each($uids))
            {
                $uid = $item['uid'];
                $exist_oss_path = 某url表->where(array('uid' => $uid))
                ->order("id desc")->field('oss_path')->find();
                if($exist_oss_path['oss_path']){
                    file_log('send_phone','已上传通讯录的uid: '.$uid,'phone');
                    continue;//跳过已经上传过内容的用户
                }
    
                $info = 某通话记录表->where(array('uid' => $uid))->field('name,number')->select();
                if(!$info||!is_array($info)){
                    file_log('send_phone','没有通讯录的uid: '.$uid,'phone');
                    continue;//跳过没有内容的用户
                }
    
                $path = 'phone/' . date("Ymd") . '/' . date('H') . '/' . $uid;
                $content = 某加密函数(json_encode(make_array_uinque($info,'name','number')),self::$contact_key_prefix.某随机字符串);
                //组装数据,并且根据name,number字段去重
                for($i=0;$i<10;$i++){
                    $result = $oss->uploadToOss($path, $content,$bucket );
                    //uploadToOss(路径,内容,$bucket="存储空间")
                    if($result){
                        break;
                    }else{
                        usleep(500);file_log('send_phone','上传oss失败uid: '.$uid." 次数".$i,'phone');
                    }
                }
                //echo $result['info']['url']."<hr>";oss返回的url
                for($i=0;$i<10;$i++){
                    $save_res = $this->save_oss_path('',$result['info']['url'],$uid);//id是pk(这里没pk),传第二,第三个参数
                    if($save_res){
                        break;
                    }else{
                        usleep(500);file_log('send_phone','修改数据库失败uid: '.$uid." 次数".$i,'phone');
                    }
                }
                if(($index>0&&$index%1000)==0){
                    $t1000 = time()- $t1;
                    file_log('send_phone','千条耗时(秒): '.$t1000,'phone');
                    $t1 = time();
                }
                if(($index>0&&$index%10000)==0){
                    $t10000 = time()- $t2;
                    file_log('send_phone','万条耗时(秒): '.$t10000,'phone');
                    $t2 = time();
                }
                //异常处理
                if (!$result||!$save_res)
                {
                    file_log('send_phone','上传失败的uid: '.$uid,'phone');
                    $que->lpush('send_phone_ids_error', $uid);
                    $false_count++;
                }else{
                    $que->incr('send_phone');//ok的数目
                    $success_count++;//成功条数
                }//异常处理完毕
                unset($uids[$index]);//悉放内存
            }
            $tt = time()-$t1;
            file_log('send_phone','结束时间,已上传总条数:'.$que->get('send_phone').
                ",本次总条数:$total_count,成功:$success_count ,失败:$false_count 消耗时间(秒):$tt,id区间$start-$end",'phone');
        }

**异常处理函数,每10分钟执行一次：**

/**
         *  保存数据和异常处理 written : yangxingyi date: 2017-09-12 11:45 Email: openweixin666@126.com
         */
        public function  handle_exception(){
            $oss = new OssCore();
            $que = 实例化redis;
            $bucket = self::$contact_bucket; // 存储空间
            $list_count = $que->llen('send_phone_ids_error');
            $time = time();
            if ($list_count > 500){
                $error_ids = $que->lrange('send_phone_ids_error', 0, -1);//返回来的一维数组
                $que->del('send_phone_ids_error');//返回来数据就干掉它
                file_log('handle_exception','大于500条异常数据的开始时间','phone');
    
                $result = $save_res = '';
                foreach($error_ids as $item)
                {
                    $uid = $item;
                    $info = 某源数据表->where(array('uid' => $uid))->field('name,number')->select();
                    if(!$info||!is_array($info)){
                        continue;//跳过没有内容的用户
                    }
                    $path = 'phone/' . date("Ymd") . '/' . date('H') . '/' . $uid;
                    $content = 加密函数(json_encode(make_array_uinque($info,'name','number')),self::$contact_key_prefix.什么鬼字符串);//组装数据
                    for($i=0;$i<10;$i++){
                        $result = $oss->uploadToOss($path, $content,$bucket );//uploadToOss(路径,内容,$bucket="存储空间")
                        if($result){
                            break;
                        }else{
                            sleep(1);
                        }
                    }
    
                    for($i=0;$i<10;$i++){
                        $save_res = $this->save_oss_path('',$result['info']['url'],$uid);
                        //id是主键索引(这没有),传递两个参数,需要传第三个参数uid
                        if($save_res){
                            break;
                        }else{
                            sleep(1);
                        }
                    }
    
                    //第二次的异常处理
                    if ($result&&$save_res)
                    {
                        $que->incr('send_phone');
                    }else{
                        file_log('handle_exception','第二次的异常处理uid: '.$uid,'phone');
                        保存错误的uid表->add(array('uid'=>$uid,'time'=>$time));
                    }
    
                }
            }
        }

**二维数组去重的方法：**

如果您读到了这里，谢谢各位看官，看完这么枯燥的文章！！！方法有不足的地方，欢迎各位大神指教！！！如果觉得对您有用的话欢迎转载，如果您有什么好文章也欢迎跟我分享！！！小弟不胜感激！！！！