一，创建别名和列出所有别名
import os
from elasticsearch import Elasticsearchpwd = os.path.dirname(os.path.realpath(__file__))ELASTICSEARCH_HOST = 'http://localhost:9200'
client = Elasticsearch(ELASTICSEARCH_HOST, basic_auth=('elastic', '1234'))# 定义别名的名字:
index_name = "product_7"
alias_name = "prod_alias_7"
# 创建别名
client.indices.put_alias(index=index_name, name=alias_name)# 列出所有的别名:# 1. 获取所有索引列表
indices = client.cat.indices(format='json')# 2. 获取所有 alias 名称
aliases = client.cat.aliases(format='json')
print(aliases)
alias_names = {alias['alias'] for alias in aliases} # 用集合方便判断
print("所有别名：")
for item in alias_names:first_char = item[0]if first_char!='.':print(item)
二，修改别名指向的索引库
import os
from elasticsearch import Elasticsearchpwd = os.path.dirname(os.path.realpath(__file__))ELASTICSEARCH_HOST = 'http://localhost:9200'
client = Elasticsearch(ELASTICSEARCH_HOST, basic_auth=('elastic', '1234'))def get_alias_all_index(alias_name):response = client.cat.aliases(name=alias_name, h='index', format='json')# 解析响应，找出所有索引indices = []for item in response:indices.append(item['index'])print('当前别名指向的索引：')print(indices)return indices# 获取别名信息
alias_name = 'prod_alias_7' # 替换为你的别名名称
new_index = "product_456"
indices = get_alias_all_index(alias_name)# 改名，让prod_alias_7指向product_456
actions = []
for old_index in indices:remove_act = {"remove": {"index": old_index, "alias": alias_name}}actions.append(remove_act)add_act = {"add": {"index": new_index, "alias": alias_name}}
actions.append(add_act)print('actions')
print(actions)
response = client.indices.update_aliases(actions=actions)# 改名完成后，查看当前别名指向的所有索引
indices = get_alias_all_index(alias_name)
三，判断别名alias是否存在？
import os
from elasticsearch import Elasticsearchpwd = os.path.dirname(os.path.realpath(__file__))ELASTICSEARCH_HOST = 'http://localhost:9200'
client = Elasticsearch(ELASTICSEARCH_HOST, basic_auth=('elastic', '1234'))# 判断alias_name是否存在
def is_exist_alias(alias_name):aliases = client.cat.aliases(format='json')# print(aliases)alias_list = []for item in aliases:first_char = item['alias'][0]if first_char != '.':print(item)alias_list.append(item['alias'])print(alias_list)# 判断是否存在于列表中if alias_name in alias_list:return Trueelse:return Falsealias_name = 'prod_alias_7'
is_alias = is_exist_alias(alias_name)
print(alias_name+"是否存在:"+str(is_alias))alias_name = 'prod_alias_a'
is_alias = is_exist_alias(alias_name)
print(alias_name+"是否存在:"+str(is_alias))
