3TB大小的ElasticSearch迁移数据量太大遇到的问题

短命女 2022-12-28 00:57 91阅读 0赞

> 本文记录最近迁移ES遇到的问题和解决办法，迁移的方式是从线上es scroll 导出到json，再想办法导入本地机房，可能迁移有其他的办法，不一定这种就是最好的，我这里选择 scroll 是因为各种限制，这样是最方便的，这里主要讨论的是scroll之后，遇到的问题。

![在这里插入图片描述][20201213110243803.png]

#### 1 scroll 拉取 ####

就是基本的 scroll ，设置了10w行做一个单独的json文件，方便我后续的处理

from utils.esd import ES_CLIENT, ES_CLIENT_LOCAL
    from elasticsearch import helpers
    import time
    import json
    import csv
    import elasticsearch
    from utils.logUtils import logger
    
    
    class GetEs:
    
        def __init__(self):
            self.espro = ES_CLIENT
            self.es_local = ES_CLIENT_LOCAL
            self.count = 0
    
        def getvalue(self):
    
            query = { 
                "query": { 
                    "match_all": { }
                }
                # , "_source": ["doc", "case_name", "case_id", "judge_date", "court_name", "case_attr"]
            }
    
            scanResp = helpers.scan(client=self.espro, query=query, size=5000, scroll="3m", index='precedents',
                                    doc_type='_doc',
                                    timeout="3m")
    
            for k in scanResp:
                yield k
    
        def insert_es(self, k):
            try:
                self.es_local.create("precedents", k["_id"].encode('utf-8'), k["_source"])
            except (elasticsearch.exceptions.ConflictError, elasticsearch.exceptions.RequestError) as e:
                logger.info("error_id" + str(k["_id"].encode('utf-8')))
                pass
    
        def write_file(self, k):
            # 这里根据这个 count 以10w行作为一个拆分
            file_index = int(self.count / 100000)
            with open('/vdd/mnt/es_out/alles-{}.json'.format(str(file_index)), 'ab') as f:
                f.write(json.dumps(k, ensure_ascii=False).encode('utf-8'))
                f.write(b'\n')
    
        def run(self):
            list_all = self.getvalue()
            for index, k in enumerate(list_all, 1):
                self.count += 1
                logger.info(self.count)
                # if self.count <= 64998:
                # continue
                self.write_file(k)
    
    
    if __name__ == "__main__":
        S = GetEs()
        S.run()

导出结果：

# 共1000个文件
    alles-655.json   alles-680.json  alles-705.json  alles-730.json  alles-756.json  alles-781.json  alles-806.json  alles-831.json  alles-857.json  alles-882.json  alles-907.json  alles-932.json  alles-958.json  alles-983.json
    .........
    .........
    
    [root@mq es_out]# ls -lh alles-757.json
    -rw-r--r-- 1 root root 2.3G 12月 10 14:39 alles-757.json

#### 2 如何将 3.3T数据导入到另一台机器的es ####

/dev/vdd       ext4      3.3T  146G    98% /vdd/mnt

因为种种愿意数据只能在机器A ，es机器在B ，A无法直接连接B，需要将数据先复制到B机器上。上面由于已经在导出的时候拆分了，所以不用再spilt ，3.6T磁盘基本也满了，3.3T文件也无法split 。

如果直接复制这 3.3T数据 到B机器，需要的时间大约是144KB/s 大约是 144天 ～  
先采取压缩

#!/bin/bash
    cd /vdd/mnt/es_out
    ls *.json > tar.log
    
    aa=$(cat tar.log|wc -l)
    
    for (( i=2;i<="$aa";i=i+1 ))
    do
        bb=$( cat tar.log|awk 'NR=='$i' {print $1}')
        echo $i
        echo $bb
        tar -jcvf  /vdd/mnt/es_tar/tares-${ i}.json.tar.bz2  /vdd/mnt/es_out/${ bb}  --remove-files
    done
    echo "successful"

这样预计可以将3.3T文件最终压缩到 220G 左右 ，再进行传输就很轻松了～  
再导完后，传输之前，再将带宽临时调整到10MB，这样下载速度大约能到1.5MB 这样

#### 3 传输之后如何导入es ####

当数据终于被传输到 B 机器，可以进行导入es了，这里也有一些道道

*  1 开始，我选择的是单条create ，这里用的是python 脚本，部分代码如下

def GetValue(self):
            with open("/home/tempData/es_out/vdd/mnt/es_out/alles-0.json", "r") as f:
                a = f.readlines()
                for i in a:
                    k = json.loads(i)
                    print(k["_id"])
                    self.insert_es(k)
    
        def insert_es(self, k):
            try:
                self.es_local.create("xxxx", k["_id"].encode('utf-8'), k["_source"])
            except (elasticsearch.exceptions.ConflictError, elasticsearch.exceptions.RequestError) as e:
                logger.info("error_id" + str(k["_id"].encode('utf-8')))
                pass

这样的写入效率大概是 200多条/s

*  2 用 bulk api 进行批量传输

def make_new_json(self):
            with open("/home/tempData/es_out/vdd/mnt/es_out/alles-111.json", "r") as f:
                a = f.readlines()
                count = 0
                actions = []
                for i in tqdm(a):
                    k = json.loads(i)
                    # 拼接插入数据结构
                    action = { 
                        "_index": "xxxx",
                        "_id": k["_id"],
                        "_source": json.dumps(k["_source"]).encode("utf-8").decode('utf-8')
                    }
                   
                    # 形成一个长度与查询结果数量相等的列表
                    actions.append(action)
                    count += 1
                    if count >= 5000:
                        helpers.bulk(ES_CLIENT_LOCAL, actions)
                            # print(a)
                            actions = []
                            count = 0
                            
                      
    
        def run(self):
            self.make_new_json()

这样批量写入大概是 1000条/s

*  3 用 bulk api + curl json 进行批量传输

将数据从scroll出来的拼成 bulk 的格式

k = json.loads(i)
    new_data = { }
    new_data['index'] = { }
    new_data['index']['_index'] = "precedents"
    new_data['index']['_id'] = str(k["_id"])
    
    # print(new_data)
    # exit(1)
    temp = json.dumps(new_data).encode("utf-8").decode('unicode_escape')
    new_jsonfile.write(temp)
    new_jsonfile.write('\n')
    temp = json.dumps(k["_source"]).encode("utf-8").decode('utf-8')
    new_jsonfile.write(temp)
    new_jsonfile.write('\n')

curl -H "Content-Type: application/json"  -XPOST "127.0.0.1:9200/precedents/_bulk" --data-binary @bulk-9.json

这样批量写入大概是 5w-10w条/s

[20201213110243803.png]: /images/20221120/0af09e9ee3f54420a768b969886b83ed.png