ES迁移方案

背景，在线迁移当前ES集群到新机器当前：ES设计了集群分片的负载平衡机制，当有新节点加入集群或者离开集群，集群会自动平衡分片的负载分布迁移目标：平滑迁移迁移策略：关闭集群自动平衡自动平衡可能会带来网络以及IO压力启动新节点与旧节点集群组成一个集群人工迁移集群数据到新节点外围访问切换到新节点关闭旧节点开启集群自动平衡迁移过程：1.配置新集群2.关闭集群自动平衡GET _cluster/settin

DBA界的小学生

1108人浏览 · 2021-04-14 14:14:36

DBA界的小学生 · 2021-04-14 14:14:36 发布

背景，在线迁移当前ES集群到新机器
当前：
在这里插入图片描述
ES设计了集群分片的负载平衡机制，当有新节点加入集群或者离开集群，集群会自动平衡分片的负载分布
迁移目标：平滑迁移
迁移策略：
关闭集群自动平衡自动平衡可能会带来网络以及IO压力
启动新节点与旧节点集群组成一个集群
人工迁移集群数据到新节点
外围访问切换到新节点
关闭旧节点
开启集群自动平衡

迁移过程：
1.配置新集群
2.关闭集群自动平衡
GET _cluster/settings
在这里插入图片描述

可以通过动态方式更新：
#禁用集群新创建索引分配禁用此选项将无法对最新的索引进行分配
cluster.routing.allocation.enable:none
#禁用集群自动平衡
cluster.routing.rebalance.enable:none
#限制索引的分布范围
“index.routing.allocation.include._ip”:“多个新集群IP”

KIBANA:

关闭自动分片
PUT _cluster/settings 
{
  "transient": {
    "cluster.routing.rebalance.enable":"none"
  }
}

PUT _cluster/settings 
{
  "transient": {
    "cluster.routing.allocation.enable":"none"
  }
}

cluster.routing.allocation.enable 设置成none，主要是影响集群中新创建的索引无法进行分片分配(把分片分配到某个节点上去)。
cluster.routing.rebalance.enable设置成none，主要是影响集群中已有索引的分片不会rebalance到(迁移)其他节点上去

可以按照ip进行排除，也可以按照节点名进行排除，在对有索引得节点操作exclude时，该节点将自动迁移索引到集群其他节点
同时存在于exclude和include时 exclude优先

PUT _cluster/settings 
{
  "transient": {
    "cluster.routing.allocation.exclude._ip": "10.10.80.59"
  }
}

PUT _cluster/settings 
{
  "transient": {
    "cluster.routing.allocation.include._ip": "10.10.80.55,10.10.80.60,10.10.80.52"
  }
}


PUT _cluster/settings 
{
  "transient": {
    "cluster.routing.allocation.exclude._name": "node-1,node-2"
  }
}

PUT _cluster/settings 
{
  "transient": {
    "cluster.routing.allocation.include._name": "node-4,node-3"
  }
}

3.启动数据节点
此时不会自动平衡分片由于cluster.routing.allocation.enable 设置成none 新建的索引将不会被分配
在这里插入图片描述

4.切换外部访问通过新节点的IP来访问ES
5.修改副本数，手动迁移数据
修改索引副本数为0，加快迁移速度
kibana：job为索引名，
PUT job/_settings
{
“number_of_replicas”: 0
}
（如果索引多也可以通过脚本修改副本数）
curl -H “Content-Type: application/json” -XPUT ‘http://10.10.80.55:9200/haha3/_settings’ -d ‘{“index”: {“number_of_replicas”: “0”}}’

手动迁移：
1.获取所有索引名：参考脚本1 1es_prepare.sh
在这里插入图片描述
2。生成迁移语句：参考脚本2 按分片迁移每个nodeX.txt都需要生成
sh 2make.sh node1.txt

执行生成得命令一个shard一个shard进行迁移（本例由node1–>node3）
sh node-1_to_node-2_action.sh >run.log

迁移前后：对比

依次迁移其他node节点过程同上
在这里插入图片描述

全部迁移后exclude旧机器node，确保旧node已无索引

6.关闭旧数据节点

7. 逐步下线旧管理节点同时启动新管理节点
先下线旧的备用主节点，添加等量的新备用主节点，最后下线正在使用的主节点
8.重新启动集群平衡
#禁用集群新创建索引分配
cluster.routing.allocation.enable:true
#禁用集群自动平衡
cluster.routing.rebalance.enable:true

副本数添加回来，迁移结束

脚本参考：

vim 1es_prepare.sh
#!/bin/bash
base_dir=/root/es_move
cd $base_dir
curl "10.10.80.55:9200/_cat/indices?pretty" |    awk '{print $3}' | grep -v '^\.'  > index_list.txt
#curl "10.10.80.55:9200/_cat/indices?pretty" |    awk '{if($2 == "open") print $3}' | grep -v '^\.'    #排除系统索引，只要open状态的索引
db_data1=`cat $base_dir/index_list.txt`
index=($db_data1)
num=${#index[@]}
i=0
1>tst
while  (($num>0))
do
echo ${index[$i]}
curl -X GET "10.10.80.55:9200/${index[$i]}/_search_shards" |jq '.shards' |sed 's/\[//g' |sed 's/\]//g' >>tst
        let i++
        let num--
done
grep -v  '^\s*$' tst >2tst  #去掉空行
cat 2tst | paste -s > tst  #把所有行统一成1行
#把tab符号替换成空格
#把空格去掉
#换行处理
sed -i 's/\t/ /g;s/[ ]*//g;s/},{/\n},{/g;s/}{/\n/g;s/},{//g;s/}{//g' tst
#此处需要手动修改
cat tst |awk -F ',' '{print $2,":",$3,":",$5,":",$6}' |sed 's/ //g'|awk -F ':' '{print $2,"*",$4,"*",$6,"*",$8}' | sed 's/ //g;s/"//g'  |sed 's/AvBEKEz5Q4iXafArWM7KXw/node-4/g;s/Spz57OmjRrCkub4l2Kg8og/node-3/g;s/5xP5tIzlRQW0uRFx61UTAg/node-2/g;s/5EouXEYlRxaCeKRaJqfwiA/node-1/g' | grep "node-1" >node1.txt
cat tst |awk -F ',' '{print $2,":",$3,":",$5,":",$6}' |sed 's/ //g'|awk -F ':' '{print $2,"*",$4,"*",$6,"*",$8}' | sed 's/ //g;s/"//g'  |sed 's/AvBEKEz5Q4iXafArWM7KXw/node-4/g;s/Spz57OmjRrCkub4l2Kg8og/node-3/g;s/5xP5tIzlRQW0uRFx61UTAg/node-2/g;s/5EouXEYlRxaCeKRaJqfwiA/node-1/g' | grep "node-2" >node2.txt
cat tst |awk -F ',' '{print $2,":",$3,":",$5,":",$6}' |sed 's/ //g'|awk -F ':' '{print $2,"*",$4,"*",$6,"*",$8}' | sed 's/ //g;s/"//g'  |sed 's/AvBEKEz5Q4iXafArWM7KXw/node-4/g;s/Spz57OmjRrCkub4l2Kg8og/node-3/g;s/5xP5tIzlRQW0uRFx61UTAg/node-2/g;s/5EouXEYlRxaCeKRaJqfwiA/node-1/g' | grep "node-3" >node3.txt
cat tst |awk -F ',' '{print $2,":",$3,":",$5,":",$6}' |sed 's/ //g'|awk -F ':' '{print $2,"*",$4,"*",$6,"*",$8}' | sed 's/ //g;s/"//g'  |sed 's/AvBEKEz5Q4iXafArWM7KXw/node-4/g;s/Spz57OmjRrCkub4l2Kg8og/node-3/g;s/5xP5tIzlRQW0uRFx61UTAg/node-2/g;s/5EouXEYlRxaCeKRaJqfwiA/node-1/g' | grep "node-4" >node4.txt

vim 2make.sh
#!/bin/bash
basedir='/root/es_move'
cd $basedir
db_data1=`cat $basedir/$1 | awk -F '*' '{print $1}'`
db_data2=`cat $basedir/$1 | awk -F '*' '{print $2}'`
db_data3=`cat $basedir/$1 | awk -F '*' '{print $3}'`
db_data4=`cat $basedir/$1 | awk -F '*' '{print $4}'`
#host=($db_data1)
node=($db_data2)
shard=($db_data3)
index_name=($db_data4)
num=${#node[@]}
i=0
#手动修改需要迁移到哪个node节点 本次迁移到node-3
aim_node='node-3'  
now_node=${node[$i]}
1>action.sh
while (($num>0))
do
        echo '索引'${index_name[$i]}'第'${shard[$i]}'分片 第'$i'次操作,还剩'$num'次' 

#生成语句：     $curl -XPOST 'http://localhost:9200/_cluster/reroute' -d '{"commands":[{"move":{"index":"filebeat-ali-hk-fd-tss1","shard":1,"from_node":"ali-hk-ops-elk1","to_node":"ali-hk-ops-elk2"}}]}'
#       echo "curl -XPOST 'http://10.10.80.55:9200/_cluster/reroute' -d '"'{"commands":[{"move":{"index":"'${index_name[$i]}'","shard":'${shard[$i]}',"from_node":"'$now_node'","to_node":"'$aim_node'"}}]}'"'"  >>action.sh
#  -H "Content-Type: application/json"
        echo "curl "'-H "Content-Type: application/json"'  " -XPOST 'http://10.10.80.55:9200/_cluster/reroute' -d '"'{"commands":[{"move":{"index":"'${index_name[$i]}'","shard":'${shard[$i]}',"from_node":"'$now_node'","to_node":"'$aim_node'"}}]}'"'"  >>action.sh
        let i++
        let num--
done
mv action.sh $now_node\_to_$aim_node\_action.sh

根据脚本生成的索引列表index_list.txt 生成取消副本脚本
vim replic.sh

#!/bin/bash
basedir='/root/es_move'
cd $basedir
db_data1=`cat $basedir/index_list.txt`
index_name=($db_data1)
num=${#index_name[@]}
i=0
1>rep.sh
while (($num>0))
do
        echo '索引'${index_name[$i]}'第'${shard[$i]}'分片 第'$i'次操作,还剩'$num'次' 
        echo "curl "'-H "Content-Type: application/json"'  " -XPUT 'http://10.10.80.55:9200/"${index_name[$i]}"/_settings' -d '"'{"index": {"number_of_replicas": "0"}}'"'"  >>rep.sh
        let i++
        let num--
done

ES导入到mysql

插入数据：
PUT /haha657/_doc/5
{
“name”:“xiao,”“he”,
“sex”:“女”,
“age”:15
}

GET /haha657/_search?
{
“took” : 805,
“timed_out” : false,
“_shards” : {
“total” : 5,
“successful” : 5,
“skipped” : 0,
“failed” : 0
},
“hits” : {
“total” : {
“value” : 4,
“relation” : “eq”
},
“max_score” : 1.0,
“hits” : [
{
“_index” : “haha657”,
“_type” : “_doc”,
“_id” : “3”,
“_score” : 1.0,
“_source” : {
“name” : “xiao,hei”,
“sex” : “女”,
“age” : 19
}
},
{
“_index” : “haha657”,
“_type” : “_doc”,
“_id” : “5”,
“_score” : 1.0,
“_source” : {
“name” : “”“xiao,”“he”"",
“sex” : “女”,
“age” : 15
}
},
{
“_index” : “haha657”,
“_type” : “_doc”,
“_id” : “2”,
“_score” : 1.0,
“_source” : {
“name” : “xiaohei”,
“sex” : “女”,
“age” : 199
}
},
{
“_index” : “haha657”,
“_type” : “_doc”,
“_id” : “1”,
“_score” : 1.0,
“_source” : {
“name” : “xiaohong”,
“sex” : “男”,
“age” : 99
}
}
]
}
}

配置logstash：
[root@slave1 /data/es/logstash-7.3.0/bin]# vim /tmp/convert_csv.conf
input{
elasticsearch {
hosts => [“10.10.80.33:9200”]
index => “haha657”
}
}
output{
csv {
fields => [“name”,“sex”,“age”]
path => “/tmp/csv.csv”
}
}

ES地址：“10.10.80.33:9200”
索引名：“haha657”
导出哪些列：fields => [“name”,“sex”,“age”]
导出位置：path => “/tmp/csv.csv”

执行导出：
[root@slave1 /data/es/logstash-7.3.0/bin]# ./logstash -f /tmp/convert_csv.conf
导出结果：
[root@slave1 /data/es/logstash-7.3.0/bin]# cat /tmp/csv.csv
xiaohong,男,99
xiaohei,女,199
“xiao,”""“he”,女,15
“xiao,hei”,女,19

mysql导入使用csv 格式导入

MariaDB [test]> load data infile ‘/tmp/tmp/aa.csv’ into table aa fields terminated by ‘,’ optionally enclosed by ‘"’ escaped by ‘"’ ;
Query OK, 4 rows affected (0.00 sec)
Records: 4 Deleted: 0 Skipped: 0 Warnings: 0

MariaDB [test]> select * from aa;
±----------±-----±-----+
| name1 | sex | age |
±----------±-----±-----+
| xiaohong | 男 | 99 |
| xiaohei | 女 | 199 |
| xiao,""he | 女 | 15 |
| xiao,hei | 女 | 19 |
±----------±-----±-----+
4 rows in set (0.00 sec)

ES清理

#!/bin/bash
for i in `cat close_index.txt`
do
#关闭索引 释放内存，可以开启
#curl -XPOST "http://host:9210/$i/_close"
#开启索引 
#curl -XPOST "http://host:9210/$i/_open"
#删除索引  释放磁盘 无法回滚
curl -XDELETE "http://host:9210/$i"
done

限速

# kibana console
PUT /_cluster/settings
{
    "persistent" : {
        "indices.recovery.max_bytes_per_sec" : "500mb"
    }
}
 
# curl
curl -X PUT "localhost:9200/_cluster/settings" -H 'Content-Type: application/json' -d'
{
    "persistent" : {
        "indices.recovery.max_bytes_per_sec" : "500mb"
    }
}
'