一个es的节点只能存一个分片的主分片或者某一个副本分片,如图所示：
分片数字相同代表为同一个分片的不同副本，当设置副本数超过节点数，则会有副本不会被分配，如下图：
存在副本未分配
移动分片命令可以将未分配的分片分配到指定节点：
commands包括以下几种：主分片move，副本allocate_replica，空分片allocate_empty_primary等分片的分配
将索引test3的分片1从2节点移动到1节点：
POST _cluster/reroute
{
"commands": [
{
"move": {
"index": "test3",
"shard": 1,
"from_node": "node-2",
"to_node": "node-1"
}
}
]
}
es默认会有分片自动迁移功能，设置集群分片不自动迁移用于实验
PUT _cluster/settings
{
"persistent": {
"cluster": {
"routing": {
"allocation.enable": "none"//all为自动迁移
}
}
}
}
rerout的Java示例：
import com.sun.deploy.util.StringUtils;
import org.elasticsearch.action.admin.cluster.reroute.ClusterRerouteRequestBuilder;
import org.elasticsearch.action.admin.cluster.reroute.ClusterRerouteResponse;
import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsGroup;
import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsResponse;
import org.elasticsearch.action.admin.cluster.stats.ClusterStatsNodeResponse;
import org.elasticsearch.action.admin.cluster.stats.ClusterStatsResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.cluster.routing.allocation.command.MoveAllocationCommand;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;
import java.util.stream.Collectors;
public class Reroute {
//es客户端的连接
private static TransportClient getEsClient(String clusterName, String ip, int port) throws UnknownHostException {
Settings settings = Settings.builder()
.put("cluster.name", clusterName)
.put("transport.tcp.connect_timeout", "120s")
.build();
TransportClient client = new PreBuiltTransportClient(settings)
.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(ip),port));
return client;
}
//获取全部数据节点备用，用于分片的分配
private static List<String> getClusterDataNodes(TransportClient client) {
ClusterStatsResponse res = client.admin().cluster().prepareClusterStats().get();
List<String> dataNodes=new ArrayList<>();
List<ClusterStatsNodeResponse> allNodes = res.getNodes();
for (ClusterStatsNodeResponse node : allNodes) {
if (node.getNode().getRoles().contains(DiscoveryNode.Role.DATA)) {
dataNodes.add(node.getNode().getName());
}
}
return dataNodes;
}
//获取每个索引的分片分布信息，用于分片配分配，节点有id好和name两个key值，一一对应
private static Map<String,List<Integer>> getIndexShardsStatus(TransportClient client,String tableName) {
Map<String,List<Integer>> name_shards=new HashMap<>();
ClusterSearchShardsResponse res2 = client.admin().cluster().prepareSearchShards(tableName).execute().actionGet();
//节点id-name对应关系
DiscoveryNode[] nodes = res2.getNodes();
Map<String,String> id_name=new HashMap<>();
for (DiscoveryNode node : nodes) {
id_name.put(node.getId(), node.getName());
}
//节点id和索引shard对应关系
ClusterSearchShardsGroup[] shards = res2.getGroups();
for (ClusterSearchShardsGroup shard : shards) {
if (!name_shards.containsKey(id_name.get(shard.getShards()[0].currentNodeId()))) {
List<Integer> ids = new ArrayList<>();
ids.add(shard.getShardId());
name_shards.put(id_name.get(shard.getShards()[0].currentNodeId()), ids);
} else {
name_shards.get(id_name.get(shard.getShards()[0].currentNodeId())).add(shard.getShardId());
}
}
return name_shards;
}
//进行不合理的分片的重新分配，适用于索引分片数小于节点数，但是某个节点却有两个主分片
public static void main(String[] args) throws UnknownHostException {
String clusterName=args[0];
String ip=args[1];
int port=Integer.parseInt(args[2]);
String tableName=args[3];
TransportClient client = getEsClient(clusterName,ip,port);
//获取dataNode
List<String> dataNodes=getClusterDataNodes(client);
System.out.println("当前集群的数据节点有 ： "+StringUtils.join(dataNodes,","));
//获取索引信息
Map<String,List<Integer>> name_shards=getIndexShardsStatus(client,tableName);
System.out.println("表《"+tableName+ "》的shards的分布情况如下：");
//找到全部的分片大于一的节点
boolean moreThanOne=false;
List<String> moreThanOneNodes=new ArrayList<>();
for (Map.Entry<String, List<Integer>> entry : name_shards.entrySet()) {
StringBuffer var2 = new StringBuffer();
for(Iterator var3 = entry.getValue().iterator(); var3.hasNext(); var2.append(var3.next())) {
if (var2.length() != 0) {
if(!moreThanOneNodes.contains(entry.getKey())){
moreThanOneNodes.add(entry.getKey());
}
moreThanOne=true;
var2.append(",");
}
}
System.out.println("=== "+entry.getKey() + "：" +var2.toString());
}
//如果有超过一个的节点且有空余的数据节点无分片则运行reroute
if( moreThanOne && dataNodes.size() > name_shards.size()){
System.out.println("需要reroute,操作如下:");
List<String> zeroNodes= dataNodes.stream().filter(num -> !name_shards.keySet().contains(num))
.collect(Collectors.toList());
ClusterRerouteRequestBuilder rerouteBulder= client.admin().cluster().prepareReroute();
int k=0;
for(int i=0;i<moreThanOneNodes.size();i++){
for(int j=1;j<name_shards.get(moreThanOneNodes.get(i)).size();j++){
int shardId=name_shards.get(moreThanOneNodes.get(i)).get(j);
rerouteBulder.add(new MoveAllocationCommand(tableName,shardId,moreThanOneNodes.get(i), zeroNodes.get(k)));
System.out.println("=== 分片"+shardId+"从"+moreThanOneNodes.get(i)+"节点移动到"+zeroNodes.get(k)+"节点");
k++;
}
}
ClusterRerouteResponse rerouteRes = rerouteBulder.execute().actionGet();
System.out.println("移动成功的分片数："+rerouteRes.getState().getRoutingNodes().getRelocatingShardCount());
}
}
}
网友评论