脚本“perf.lua” local random = math.random local reqs = {} local cnt = 0 -- 压测的查询请求文件名称根据需要调整。 for line in io.lines("requests.txt") do table.insert(reqs, line) cnt = cnt + 1 end local addrs = {} local counter = 0 function setup(thread) local append = function(host, port) for i, addr in ipairs(wrk.lookup(host, port)) do if wrk.connect(addr) then addrs[#addrs+1] = addr end end end if #addrs == 0 then -- 根据集群的实际地址进行修改。 append("x.x.x.x", 9200) append("x.x.x.x", 9200) append("x.x.x.x", 9200) end local index = counter % #addrs + 1 counter = counter + 1 thread.addr = addrs[index] end -- 索引名称根据需要调整。 wrk.path = "/index_sift_graph/_search?request_cache=false&preference=_local" wrk.method = "GET" wrk.headers["Content-Type"] = "application/json" function request() return wrk.format(wrk.method, wrk.path, wrk.headers, reqs[random(cnt)]) end

测试前准备 创建Elasticsearch向量数据库，参考创建Elasticsearch集群。 “节点数量”选择“3”，“节点规格”选择“通用计算型”的“4vCPUs | 16GB”（由于测试的数据量不大，且为了和第三方的基线测试保持相同的CPU规格），“节点存储”选择“超高I/O”，不启用安全模式。 获取测试数据集。 sift-128-euclidean：维度128，base数据100万条，使用欧式距离度量。 cohere-768-cosine：维度768，base数据100万条，使用余弦距离度量。 gist-960-euclidean：维度960，base数据100万条，使用欧式距离度量。 “sift-128-euclidean”和“gist-960-euclidean”数据的下载地址是https://github.com/erikbern/ann-benchmarks。如需使用“cohere-768-cosine”数据，请提交工单获取。 图1 下载“sift-128-euclidean”和“gist-960-euclidean”数据 准备测试工具。 准备数据写入和召回率测试脚本，参考脚本base_test_example.py。 下载性能测试使用的开源压测工具Wrk，获取地址https://github.com/wg/wrk/tree/master。

性能测试的操作步骤 创建一个弹性 云服务器ECS ，用于安装压测工具和执行测试脚本。操作指导请参见快速购买和使用Linux E CS 。 ECS必须和Elasticsearch集群在同一个虚拟私有云和安全组中。 也可以使用其他客户端服务器，但是必须保证服务器和Elasticsearch集群在同一VPC。 将测试数据集上传到ECS上。 将数据写入和召回率测试脚本上传到ECS上，并执行如下命令。 pip install h5py pip install elasticsearch==7.10 python3 base_test_example.py 执行完成后，会创建测试的向量索引，写入测试数据，并返回平均查询召回率Recall。 在ECS上安装开源压测工具Wrk。 在ECS上准备压测的查询请求文件，用于模拟真实业务场景。参考脚本prepare_query.py。 pip install h5py python3 prepare_query.py 在ECS上准备Wrk的压测配置脚本。参考脚本perf.lua，脚本中的查询请求文件名称、集群访问地址和索引名称需要根据实际环境修改。 在ECS执行如下命令进行向量检索的性能压测。 wrk -c60 -t60 -d10m -s perf.lua http://x.x.x.x:9200 “t”表示压测线程数。 “c”表示与服务端的连接数。 “d”表示压测时间，“10m”表示10分钟 。 “s”表示Wrk的压测配置脚本。 “x.x.x.x”表示Elasticsearch集群的访问地址。 在回显中获得测试数据，其中“Requests/sec”即查询吞吐量QPS。 图2 测试结果示例

性能测试比较 GRAPH类索引 百万规模的场景推荐使用GRAPH索引类型。 测试方案一：使用不同维度的数据集，在Top10召回率均达到99%的情况下，测试向量数据库能支撑的最大QPS。每个数据集均基于默认参数和调优参数分别进行测试，通过调整构建参数可以使得图索引结构更优，在同等召回精度下能取得更高的查询性能。 测试结果： 表1 GRAPH类索引测试结果1 数据集 构建参数 查询参数 性能指标 efc shrink ef max_scan_num QPS Recall sift-128-euclidean 200 1.0 84 10000 15562 0.99 500 0.8 50 10000 17332 0.99 cohere-768-cosine 200 1.0 154 10000 3232 0.99 500 0.95 106 10000 3821 0.99 gist-960-euclidean 200 1.0 800 19000 860 0.99 500 0.9 400 15000 1236 0.99 结论：对于不同的数据集，使用默认参数均能达到99%以上的召回率。在进一步调整构建参数和查询参数后，增加了一定的索引构建开销，同时也达到更高的查询性能。 测试方案二：使用同一数据集，通过调整索引参数，测试不同召回率下的查询性能。本方案用COHERE数据集，分别测试了Top10召回率为99%、98%及95%时的集群最大QPS。 测试结果： 表2 GRAPH类索引测试结果1 数据集 构建参数 查询参数 性能指标 efc ef QPS Recall cohere-768-cosine 500 128 3687 0.99 500 80 5320 0.98 500 36 9028 0.95 结论：同一集群在统一索引构建参数的情况下，通过调整ef参数可以获得不同的查询精度，在略微损失召回率的场景下可以获得成倍的性能提升。 GRAPH_PQ类索引 基于图算法的索引为了保证查询性能通常需要常驻内存，因此当向量维度较高或数据量较大时，内存资源成为影响成本及性能的关键因素。具体来说，高维度的向量和大规模的数据集对内存的需求显著增加，这不仅关系到存储成本，还直接影响到索引算法的运行效率和响应速度。该场景推荐使用GRAPH_PQ索引类型。 测试方案：使用维度较高的COHERE与GIST数据集，测试在Top10召回率达到95%时的集群最大QPS，并与GRAPH索引对比常驻内存开销。 测试结果： 表3 GRAPH_PQ类索引测试结果 数据集 构建参数 查询参数 性能指标 内存开销 efc fragment_num ef topk QPS Recall GRAPH_PQ GRAPH cohere-768-cosine 200 64 85 130 8723 0.95 332MB 3.3GB gist-960-euclidean 200 120 200 360 4267 0.95 387MB 4.0GB 结论：结果显示使用GRAPH_PQ类索引能够在节约10倍+内存开销的情况下，取得与GRAPH索引差不多的精度和性能。因此， CSS 向量索引的GRAPH_PQ算法融合了图索引与量化算法，能够大幅降低内存的开销，提升单机的数据容量。 测试数据中涉及的索引参数说明请参见表4，关于构建参数的详细说明请参见在Elasticsearch集群创建向量索引，关于查询参数的详细说明请参见在Elasticsearch集群使用向量索引搜索数据。 表4 索引参数说明 类型 参数名称 说明 构建参数 efc 构建hnsw时考察邻居节点的队列大小，默认值为200，值越大精度越高，构建速度将会变慢。 shrink 构建hnsw时的裁边系数，默认值为1.0f。 fragment_num 段数，默认值为0，插件自动根据向量长度设置合适的段数。 查询参数 ef 查询时考察邻居节点的队列大小。值越大查询精度越高，查询速度会变慢。默认值为200。 max_scan_num 扫描节点上限。值越大精度越高，查询速度变慢。默认值为10000。 topk 查询时返回top k条数据。

脚本“base_test_example.py” # -*- coding: UTF-8 -*- import json import time import h5py from elasticsearch import Elasticsearch from elasticsearch import helpers def get_client(hosts: list, user: str = None, password: str = None): if user and password: return Elasticsearch(hosts, http_auth=(user, password), verify_certs=False, ssl_show_warn=False) else: return Elasticsearch(hosts) # 索引参数说明请参见在Elasticsearch集群创建向量索引。 def create(es_client, index_name, shards, replicas, dim, algorithm="GRAPH", metric="euclidean", neighbors=64, efc=200, shrink=1.0): index_mapping = { "settings": { "index": { "vector": True }, "number_of_shards": shards, "number_of_replicas": replicas, }, "mappings": { "properties": { "id": { "type": "integer" }, "vec": { "type": "vector", "indexing": True, "dimension": dim, "algorithm": algorithm, "metric": metric, "neighbors": neighbors, "efc": efc, "shrink": shrink, } } } } es_client.indices.create(index=index_name, body=index_mapping) print(f"Create index success! Index name: {index_name}") def write(es_client, index_name, vectors, bulk_size=1000): print("Start write! Index name: " + index_name) start = time.time() for i in range(0, len(vectors), bulk_size): actions = [{ "_index": index_name, "id": i + j, "vec": v.tolist() } for j, v in enumerate(vectors[i: i + bulk_size])] helpers.bulk(es_client, actions, request_timeout=180) print(f"Write success! Docs count: {len(vectors)}, total cost: {time.time() - start:.2f} seconds") merge(es_client, index_name) def merge(es_client, index_name, seg_cnt=1): print(f"Start merge! Index name: {index_name}") start = time.time() es_client.indices.forcemerge(index=index_name, max_num_segments=seg_cnt, request_timeout=7200) print(f"Merge success! Total cost: {time.time() - start:.2f} seconds") # 查询参数说明请参考见在Elasticsearch集群使用向量索引搜索数据。 def query(es_client, index_name, queries, gts, size=10, k=10, ef=200, msn=10000): print("Start query! Index name: " + index_name) i = 0 precision = [] for vec in queries: hits = set() dsl = { "size": size, "stored_fields": ["_none_"], "docvalue_fields": ["id"], "query": { "vector": { "vec": { "vector": vec.tolist(), "topk": k, "ef": ef, "max_scan_num": msn } } } } res = es_client.search(index=index_name, body=json.dumps(dsl)) for hit in res['hits']['hits']: hits.add(int(hit['fields']['id'][0])) precision.append(len(hits.intersection(set(gts[i, :size]))) / size) i += 1 print(f"Query complete! Average precision: {sum(precision) / len(precision)}") def load_test_data(src): hdf5_file = h5py.File(src, "r") base_vectors = hdf5_file["train"] query_vectors = hdf5_file["test"] ground_truths = hdf5_file["neighbors"] return base_vectors, query_vectors, ground_truths def test_sift(es_client): index_name = "index_sift_graph" vectors, queries, gts = load_test_data(r"sift-128-euclidean.hdf5") # 根据实际测试需求调整分片和副本数、索引算法、索引参数等。本文性能测试均配置的是1个分片、2个副本。 create(es_client, index_name, shards=1, replicas=2, dim=128) write(es_client, index_name, vectors) query(es_client, index_name, queries, gts) if __name__ == "__main__": # 此处修改为CSS集群的实际访问地址。 client = get_client(['http://x.x.x.x:9200']) test_sift(client)

写入性能优化 基于Elasticsearch的数据写入流程分析，有以下几种性能优化方案。 表1 写入性能优化 优化方案 方案说明 使用SSD盘或升级集群配置 使用SSD盘可以大幅提升数据写入与merge操作的速度，对应到CSS服务，建议选择“超高IO型”存储，或者超高IO型主机。 采用Bulk API 客户端采用批量数据的写入方式，每次批量写入的数据建议在1~10MB之间。 随机生成_id 如果采用指定_id的写入方式，数据写入时会先触发一次查询操作，进而影响数据写入性能。对于不需要通过_id检索数据的场景，建议使用随机生成的_id。 设置合适的分片数 分片数建议设置为集群数据节点的倍数，且分片的大小控制在50GB以内。 关闭副本 数据写入与查询错峰执行，在数据写入时关闭数据副本，待数据写入完成后再开启副本。 Elasticsearch 7.x版本中关闭副本的命令如下： PUT {index}/_settings { "number_of_replicas": 0 } 调整索引的刷新频率 数据批量写入时，可以将索引的刷新频率“refresh_interval”设置为更大的值或者设置为“-1”（表示不刷新），通过减少分片刷新次数提高写入性能。 Elasticsearch 7.x版本中，将更新时间设置为15s的命令如下： PUT {index}/_settings { "refresh_interval": "15s" } 优化写入线程数与写入队列大小 为应对突发流量，可以适当地提升写入线程数与写入队列的大小，防止突发流量导致出现错误状态码为429的情况。 Elasticsearch 7.x版本中，可以修改如下自定义参数实现写入优化：thread_pool.write.size，thread_pool.write.queue_size。 设置合适的字段类型 指定集群中各字段的类型，防止Elasticsearch默认将字段猜测为keyword和text的组合类型，增加不必要的数据量。其中keyword用于关键词搜索，text用于全文搜索。 对于不需要索引的字段，建议“index”设置为“false”。 Elasticsearch 7.x版本中，将字段“field1”设置为不建构索引的命令如下： PUT {index} { "mappings": { "properties": { "field1":{ "type": "text", "index": false } } } } 优化shard均衡策略 Elasticsearch默认采用基于磁盘容量大小的Load balance策略，在多节点场景下，尤其是在新扩容的节点上，可能出现shard在各节点上分配不均的问题。为避免这类问题，可以通过设置索引级别的参数“routing.allocation.total_shards_per_node”控制索引分片在各节点的分布情况。此参数可以在索引模板中配置，也可以修改已有索引的setting生效。 修改已有索引的setting的命令如下： PUT {index}/_settings { "index": { "routing.allocation.total_shards_per_node": 2 } }

数据写入流程 图1 数据写入流程 如图1所示，以Elasticsearch集群为例，介绍客户端往Elasticsearch或OpenSearch集群中写入数据的流程。图中的P表示主分片Primary，R表示副本分片Replica，主副分片在数据节点Node里是随机分配的，但是不能在同一个节点里。 客户端向Node1发送写数据请求，此时Node1为协调节点。 节点Node1根据数据的_id将数据路由到分片2，此时请求会被转发到Node3，并执行写操作。 当主分片写入成功后，它将请求转发到Node2的副本分片上。当副本写入成功后，Node3将向协调节点报告写入成功，协调节点向客户端报告写入成功。 Elasticsearch中的单个索引由一个或多个分片(shard)组成，每个分片包含多个段（Segment），每一个Segment都是一个倒排索引。 图2 Elasticsearch的索引组成 如图3所示，将文档插入Elasticsearch时，文档首先会被写入缓冲区Buffer中，同时写入日志Translog中，然后在刷新时定期从该缓冲区刷新文档到Segment中。刷新频率由refresh_interval参数控制，默认每1秒刷新一次。更多写入性能相关的介绍请参见Elasticsearch的官方介绍Near Real-Time Search。 图3 文档插入Elasticsearch的流程

解决方案 方案一：关闭或者删除不用的索引，减少shard数量。 方案二：修改节点的shard数量的限制，参数配置请参考max_shards_per_node。 PUT _cluster/settings{ "persistent": { "cluster": { "max_shards_per_node": 2000 } }} 修改节点的shard数量的限制属于临时规避方案，如果要长期解决，建议每GB的JVM堆内存小于等于20个shards（节点堆内存大小为节点内存规格的1/2，最大值为31GB），shard的建议数量详细请参见shard count recommendation。

集群一直处于快照中 集群一直处于快照中，有三个比较常见的原因： 集群数据量大或者集群压力大，备份快照耗时长。 单个节点的快照速度默认是40MB/s，同时，快照的性能还受集群情况影响，如果此时集群负载较高，耗时将会更久。可以通过上述章节的查询单个快照信息查询正在执行的快照情况。 执行GET _snapshot/repo_auto/snapshot-name，可以看到剩余还需要完成的shard个数，也可以通过删除快照接口提前终止。 解决方法：等待或者提前终止。 快照信息更新失败。 Elasticsearch将进行中的快照信息保存在cluster state中，快照完成后需要更新快照状态，由于Elasticsearch更新快照状态的接口没有加入重试或者容错机制，比如由于当时集群内存压力大，更新快照动作被熔断，那么这个快照将会一直处于快照中。 解决方法：调用快照删除接口。 临时AK、SK过期。 CSS通过委托将Elasticsearch中的数据写入到用户的OBS中，快照仓库创建的时候，需要去使用委托获取临时的AK 、SK设置到仓库中。由于临时的AK、SK是有时效性的（24小时过期），如果一个快照超过24小时还未完成，那么这个快照将会失败。这种情况会有一个比较大的风险，因为此时仓库的AK、SK过期，无法对这个仓库进行更新、查询、删除操作，这种情况下cluster state信息将会无法清除，只能通过普通重启（滚动重启无法生效）集群来清除cluster state里面残留的快照信息。 解决方法：暂时只能通过普通重启集群来消除，后期CSS会提供终止接口，可以来解决这种无法消除状态的现象。 父主题： 功能使用类

排查是否有权限 登录 统一身份认证 服务管理控制台。 查看当前登录所用的账号或 IAM 用户所属的用户组。 具体操作请参见《统一身份认证服务用户指南》中的查看或修改用户信息章节。 查看用户组的权限中是否包含：“全局服务”中“ 对象存储服务 ”项目的“Tenant Administrator”权限、当前所属区域的“Elasticsearch Administrator”权限。 具体操作请参见《统一身份认证服务用户指南》中的查看或修改用户组章节。 如果用户组的权限中不包含以上两个权限，请执行4。 如果用户组的权限中包含以上两个权限，请联系人工客服协助解决。 为用户组添加：“全局服务”中“对象存储服务”项目的“Tenant Administrator”权限、当前所属区域的“Elasticsearch Administrator”权限。 具体操作请参见《统一身份认证服务用户指南》中的查看或修改用户组章节。

问题现象 ECS服务器部署logstash，然后推送数据到 云搜索服务 CSS，出现错误信息如下： LogStash::Outputs::ElasticSearch::HttpClient::Pool::BadResponseCodeError: Got response code '500' contacting Elasticsearch at URL 'https://192.168.xx.xx:9200/_xpack'。

为什么集群创建失败 集群创建失败原因有如下4种： 资源配额不足，无法创建集群。建议申请足够的资源配额，详情请参见如何申请扩大配额？。 如果集群配置信息中，“安全组”的“端口范围/ICMP类型”不包含“9200”端口，导致集群创建失败。请修改安全组信息或选择其他可用安全组。 7.6.2以及7.6.2之后的版本，集群内通信端口9300默认开放在用户VPC的子网上面。创建集群时需要确认所选安全组是否放通子网内的9300通信端口，如果未放通，请修改安全组信息或选择其他可用安全组。 权限不足导致集群创建失败。建议参考权限管理获取权限，然后创建集群。 父主题： 访问集群类

处理步骤 在集群管理页面，单击不可用的集群名称，进入集群基本信息页面。 单击“配置信息”中的安全组名称，进入当前集群所选安全组的基本信息页面。 分别查看“入方向规则”和“出方向规则”页签下，是否存在“策略”为“允许”，“协议端口”为“TCP : 9300”，“类型”为“IPv4”的安全组规则。 是，联系技术支持定位集群不可用问题。 否，执行下一步。 修改集群当前所选安全组信息，放通9300通信端口。 在当前集群所选安全组基本信息界面，选择“入方向规则”页签。 单击“添加规则”，在添加入方向规则对话框设置“优先级”为“100”，“策略”选择“允许”，“协议端口”选择“基本协议/自定义TCP”，端口填写“9300”，“类型”选择“IPv4”，“源地址”选择“安全组”下的集群当前安全组名称，即同安全组内放通。 图2 添加安全组规则 单击“确定”即可完成放通9300端口的设置。 同样的步骤，在“出方向规则”页签添加放通9300端口的设置。 安全组放通9300端口后，等待集群自动恢复可用状态。

问题现象 安装自定义插件后重启集群，“集群状态”变为“不可用”。 单击集群名称进入集群基本信息页面，选择“日志管理”，单击“日志查询”页签，可见日志内容存在明显的关于插件的报错“fatal error in thread [main], exitingjava.lang. NoClassDefFoundError: xxx/xxx/.../xxxPlugin at ...”。 图1 节点报错日志示例 CSS服务已下线自定义插件功能，但历史版本的集群可能还装有自定义插件，只有这类集群可能出现该故障。

问题现象 “集群状态”为“不可用”。 单击集群名称进入集群基本信息页面，选择“日志管理”，单击“日志查询”页签，可见日志内容存在警告“master not discovered or elected yet, an election requires at least 2 nodes with ids [xxx, xxx, xxx, ...], have discovered [xxx...] which is not a quorum”。 图1 节点报错日志示例