Easyearch 为了防止索引将磁盘空间完全占满，使用磁盘水位线进行磁盘空间控制。之前有文章提过不同水位线的作用，以及如何使用 INFINI Console 提前进行告警，提前进行处理。本篇主要探讨提前处理的情况。

一、增加资源

如果资源充裕，可考虑为 Easysearch 集群扩充资源：

1. 添加新的数据节点

扩充节点后，集群会自动进行数据平衡，可用下面的命令查看进度

GET /_cat/shards?v&h=state,node&s=state

如果响应中分片的状态是 RELOCATING ，则表示分片仍在移动。

2. 扩充现有数据节点磁盘容量

扩充完后可查看磁盘利用率下降情况

GET _cat/allocation?v&s=disk.avail&h=node,disk.percent,disk.avail,disk.total,disk.used,disk.indices,shards

二、释放磁盘空间

如果无资源可添加，则考虑减少磁盘消耗：

1. 删除无用索引

建议使用索引生命周期进行管理，自动删除过期索引。

2. 删除多余副本

有些业务索引可能会有多分副本，可酌情缩减副本数，降低磁盘消耗。以下命令按副本数量和主存储大小的降序排列索引。

GET _cat/indices?v&s=rep:desc,pri.store.size:desc&h=health,index,pri,rep,store.size,pri.store.size

3. 可搜索快照

对于有些数据平时不常用，但需要长期保留的，建议使用可搜索快照功能降低磁盘消耗。

三、索引空间优化

1. 启用 ZSTD 压缩及 source_reuse 功能

Easysearch 支持 ZSTD 和 source_reuse 功能，对比默认的压缩算法，可大幅降低磁盘消耗。

可在创建索引时启用 ZSTD 和 source_reuse 功能，也可通过索引模板来进行设置，参考文档。

PUT test-index
{
  "settings": {
    "index.codec": "ZSTD",
    "index.source_reuse": "true"
  }
}

⚠️ 注意：当索引里包含 nested 类型映射，或插件额外提供的数据类型时，不能启用 source_reuse，例如 knn 索引。

2. 索引优化

• mapping 优化
  避免使用默认的 mapping 类型，因为字符串类型的数据将得到 text 和 keyword 两个类型的 mapping。
• 字段优化
  统计指定索引每个字段的访问次数。

GET metrics/_field_usage_stats

分析指定索引各个字段占用磁盘的大小。

POST metrics/_disk_usage?run_expensive_tasks=true

结合以上信息进一步优化各个字段，如关闭不用的功能等

3. 使用 rollup 功能

对于时序场景类的数据，往往会有大量的非常详细的聚合指标，随着时间的图推移，存储将持续增长。汇总功能可以将旧的、细粒度的数据汇总为粗粒度格式以进行长期存储。通过将数据汇总到一个单一的文档中，可以大大降低历史数据的存储成本。

Easysearch 的 rollup 具备一些独特的优势，可以自动对 rollup 索引进行滚动而不用依赖其他 API 去单独设置，并且在进行聚合查询时支持直接搜索原始索引，做到了对业务端的搜索代码完全兼容，从而对用户无感知。

如果有问题，欢迎加我微信沟通。

关于 Easysearch

INFINI Easysearch 是一个分布式的搜索型数据库，实现非结构化数据检索、全文检索、向量检索、地理位置信息查询、组合索引查询、多语种支持、聚合分析等。Easysearch 可以完美替代 Elasticsearch，同时添加和完善多项企业级功能。Easysearch 助您拥有简洁、高效、易用的搜索体验。

官网文档：https://docs.infinilabs.com/easysearch

作者：杨帆，极限科技（INFINI Labs）高级解决方案架构师、《老杨玩搜索》栏目 B 站 UP 主，拥有十余年金融行业服务工作经验，熟悉 Linux、数据库、网络等领域。目前主要从事 Easysearch、Elasticsearch 等搜索引擎的技术支持工作，服务国内私有化部署的客户。

去年我们尝试过使用 Easysearch + 千问 2 大模型打造一个企业内部知识问答系统，今年又有更加给力的大模型出现了--DeepSeek，性能对标 OpenAI o1 正式版。而且 Easysearch 对比去年也有了不少进步，是时候让我们升级下问答系统了。

DeepSeek

2025 年 1 月 20 日，人工智能领域迎来里程碑式突破！深度求索（DeepSeek）正式发布新一代推理大模型 DeepSeek-R1，不仅实现与 OpenAI 最新 o1 正式版的性能对标，更以全栈开放的生态布局引发行业震动。DeepSeek-R1 是首个遵循 MIT License 开源协议的高性能推理模型，完全开源，不限制商用，无需申请，极大地推动了 AI 技术的开放与共享。

下载模型

我们使用 ollama 下载运行 DeepSeek-R1，根据本地资源情况选择一个大小合适的版本：8b。

• 8b 蒸馏模型源自 Llama3.1-8B-Base
• 7b 蒸馏模型源自 Qwen-2.5 系列

这两个可能是个人用户使用最多的选择，大家资源充足的可以都下载下来对比下效果。

由于是升级，我们只需在原有程序基础上替换新版本的 Easysearch 和集成 DeepSeek 即可，Easysearch 升级成新版本 1.10.1，程序框架和 embedding 模型 (mxbai-embed-large:latest) 仍然保持不变。

数据准备

跟上次一样，使用 "INFINI 产品安装手册.PDF" 作为知识内容，通过程序将文档内容切片、转换成向量后写入 Easysearch 存储，然后结合大模型对其中的内容进行提问。

程序调整

程序代码需要调整 LLM 为 deepseek-r1:8b。另外本地主机资源有限，为节约时间，取消上个版本的用户问题改写功能(注释部分)。定义新的 retriever 和 qa_chain 直接将用户问题和 context 信息发送给大模型。

# # 实例化一个大模型工具
from langchain_community.chat_models import ChatOllama

llm = ChatOllama(model="deepseek-r1:8b")

# from langchain.prompts import PromptTemplate

# my_template = PromptTemplate(
#     input_variables=["question"],
#     template="""You are an AI language model assistant. Your task is
#     to generate 3 different versions of the given user
#     question in Chinese to retrieve relevant documents from a vector  database.
#     By generating multiple perspectives on the user question,
#     your goal is to help the user overcome some of the limitations
#     of distance-based similarity search. Provide these alternative
#     questions separated by newlines. Original question: {question}""",
# )

# 实例化一个MultiQueryRetriever
# retriever_from_llm = MultiQueryRetriever.from_llm(
#     retriever=docsearch.as_retriever(),
#     llm=llm,
#     prompt=my_template,
#     include_original=True)

retriever = docsearch.as_retriever()
# 实例化一个RetrievalQA链
qa_chain = RetrievalQA.from_chain_type(llm, retriever=retriever)

至此程序修改已经完成，原程序算上注释也不过 100 来行，大家感兴趣的可以去查看原博客。

效果测试

模拟用户提问：网关运行后监听哪个端口。

系统回答如下。

在回答中，可以看到 DeepSeek 的"思考"过程，另外回答结果也非常正确，文档中原文还是用的英语 INFINI Gateway 表示网关。

模拟用户提问：LOGGING_ES_ENDPOINT 有什么用。

系统回答如下。

文档原文内容如下。

好了，我对 DeepSeek 的表现很满意，至此知识问答系统就升级完了。

如有任何问题，请随时联系我，期待与您交流！

关于 Easysearch

INFINI Easysearch 是一个分布式的搜索型数据库，实现非结构化数据检索、全文检索、向量检索、地理位置信息查询、组合索引查询、多语种支持、聚合分析等。Easysearch 可以完美替代 Elasticsearch，同时添加和完善多项企业级功能。Easysearch 助您拥有简洁、高效、易用的搜索体验。

官网文档：https://infinilabs.cn/docs/latest/easysearch

作者：杨帆，极限科技（INFINI Labs）高级解决方案架构师、《老杨玩搜索》栏目 B 站 UP 主，拥有十余年金融行业服务工作经验，熟悉 Linux、数据库、网络等领域。目前主要从事 Easysearch、Elasticsearch 等搜索引擎的技术支持工作，服务国内私有化部署的客户。

Easyearch 为了防止索引将磁盘空间完全占满，使用磁盘水位线进行磁盘空间控制。具体来说有三条磁盘水位线：low、high、flood。

低水位线

通过参数 cluster.routing.allocation.disk.watermark.low 进行设置，默认值 85%。也可设置成一个具体值，比如：400mb，代表须保留 400mb 空闲磁盘空间，否则就算超水位线。

一旦节点磁盘使用率超过了低水位线，Easysearch 集群不会将分片分配至该节点，但是不影响新建索引的主分片分配到该节点，新建索引的副本分配不能分配到该节点。

如果所有节点都超过高水位线，此时创建新索引会导致集群状态变成 yellow。

高水位线

通过参数 cluster.routing.allocation.disk.watermark.high 进行设置，默认值 90%。也可设置成一个具体值，比如：300mb，代表须保留 300mb 空闲磁盘空间，否则就算超水位线。

一旦节点磁盘使用率超过了高水位线，Easysearch 集群会尝试将分片移动到其他节点，不允许任何分片分配到该节点。

如果所有节点都超过高水位线，此时创建新索引会导致集群状态变成 red。

洪水位线

通过参数 cluster.routing.allocation.disk.watermark.flood_stage 进行设置，默认值 95%。也可设置成一个具体值，比如：200mb，代表须保留 200mb 空闲磁盘空间，否则就算超水位线。

一旦节点磁盘使用率超过了洪水位线，Easysearch 集群会为该节点上的所有索引添加只读锁，包括系统索引。只读锁会阻止新数据写入，当磁盘利用率低于高水位线时，只读锁会自动释放。

针对节点磁盘使用率，我们可以使用 INFINI Console 进行节点磁盘使用率告警，便于我们及时发现问题苗头，提前进行处理。有任何问题，欢迎加我微信沟通。

关于 Easysearch

INFINI Easysearch 是一个分布式的搜索型数据库，实现非结构化数据检索、全文检索、向量检索、地理位置信息查询、组合索引查询、多语种支持、聚合分析等。Easysearch 可以完美替代 Elasticsearch，同时添加和完善多项企业级功能。Easysearch 助您拥有简洁、高效、易用的搜索体验。

官网文档：https://infinilabs.cn/docs/latest/easysearch

作者：杨帆，极限科技（INFINI Labs）高级解决方案架构师、《老杨玩搜索》栏目 B 站 UP 主，拥有十余年金融行业服务工作经验，熟悉 Linux、数据库、网络等领域。目前主要从事 Easysearch、Elasticsearch 等搜索引擎的技术支持工作，服务国内私有化部署的客户。

DataX

DataX 是阿里开源的一款离线数据同步工具，致力于实现包括关系型数据库(MySQL、Oracle 等)、HDFS、Hive、ODPS、HBase、FTP 等各种异构数据源之间稳定高效的数据同步功能。

本篇主要介绍 DataX 如何将数据写入到 Easysearch，对于各种数据源的连接不会做深入的探讨，感兴趣的小伙伴可以访问 DataX 的 Github 仓库查看详情。

下载与安装

DataX 无需安装，下载后解压即可使用。

系统需求：

1. JDK 1.8 及以上
2. Python2 或 3

创建任务配置文件

每个数据同步的操作可称为一个任务，任务的配置文件定义了数据源(reader)、数据目的(writer) ，以及任务的设置信息，如并发数、速度控制等。DataX 集成了如此多的数据源，如果靠纯手工编写任务配置显然不现实。官方也出了个命令可以根据指定的数据源和数据目的帮助大家生成任务配置。

python datax.py -r {YOUR_READER} -w {YOUR_WRITER}

测试配置文件

此次演示使用 streamreader 和 elasticsearchwriter 作为数据源和数据目的，任务配置如下：

{
  "job": {
    "content": [
      {
        "reader": {
          "name": "streamreader",
          "parameter": {
            "sliceRecordCount": 10000,
            "column": [
              {
                "type": "long",
                "value": "10"
              },
              {
                "type": "string",
                "value": "hello，你好，世界-DataX"
              },
              {
                "type": "string",
                "value": "hello，你好，Easysearch"
              }
            ]
          }
        },
        "writer": {
          "name": "elasticsearchwriter",
          "parameter": {
            "endpoint": "http://localhost:9200",
            "accessId": "admin",
            "accessKey": "1ef0c661d8562aaa06be",
            "index": "yf-test",
            "column": [
              { "name": "no", "type": "long" },
              { "name": "content", "type": "keyword" },
              { "name": "content2", "type": "keyword" }
            ]
          }
        }
      }
    ],
    "setting": {
      "speed": {
        "channel": 50
      }
    }
  }
}

streamreader 是一个从内存读取数据的插件， 它主要用来快速生成期望的数据并对写入插件进行测试。

我们用 streamreader 构造了 10000 个文档，文档含三个字段，任务启动了 50 个 channel 进行数据发送，结果就是共计发送 50w 个文档。

elasticssearchwriter 指定了 Easysearch 的连接信息：

• endpoint: Easysearch 的地址和端口
• accessId: 用户名
• accessKey: 密码
• index: 写入索引名
• column: 对 reader 发来数据的 schema 定义
• batchsize: 默认 1000

这次我们 Easysearch 开启的 http 服务，因为 DataX 的 elasticsearchwriter 无法跳过证书验证。对于必须使用 https 的场景，可使用 INFINI Gateway 代理 ES 服务，提供 http 通道给离线数据同步专用。

⚠️ 注意：

不同的 reader、writer 对 sliceRecordCount 和 channel 会有不同的行为。

Easysearch

本次测试使用的 Easysearch 版本是 1.9.0，需要注意是 Easysearch 要开启兼容性参数：

elasticsearch.api_compatibility: true

否则创建索引报错退出。(实际索引创建成功了但是 mapping 信息是空的)

运行任务

编辑好任务配置文件后，下一步就是执行任务。

python3 datax.py yf-test.json

写入数据时索引不存在，Datax 根据 schema 定义创建了索引。

OK 任务执行完毕，写入 50w 个文档耗时 10 秒。

如果有其他问题欢迎与我联系。

本文将探讨如何使用 INFINI Gateway 阻止不合理的查询发送到 Elasticsearch，此方法同样适用于 Opensearch 和 Easysearch 。

在以往处理 Elasticsearch OOM（内存溢出）问题的经验中，我们发现许多案例是由于查询操作导致节点出现 OOM 的情况。经过调查，这些案例主要分为两类：一类是由于查询吞吐量超出了集群的处理能力，另一类则是在执行某些不合理的查询时触发了 OOM。

具体来说：

• 查询吞吐量过高：当查询请求的频率或复杂度超过了集群的处理能力时，可能会导致节点内存耗尽，从而引发 OOM。
• 执行不合理查询：某些特殊类型的查询（例如涉及大量嵌套、深度分页或复杂的聚合操作）可能需要大量的内存资源，在执行过程中也可能导致 OOM。

通过识别并优化这些查询模式，可以有效减少 OOM 事件的发生。针对查询吞吐量过高的情况，可以参考之前的文章来管理查询吞吐。接下来的内容将介绍如何阻挡不合理查询，保护集群稳定。

不合理查询

不合理查询是指那些消耗过多系统资源（如 CPU、内存）、设计复杂、执行时间过长或需要大量计算资源的查询。这类查询不仅会导致高负载和资源耗尽，影响整个集群的稳定性和响应速度，还可能对用户体验产生负面影响。

典型的不合理查询包括但不限于：

• 嵌套聚合查询
• 使用复杂的正则表达式进行模糊匹配
• 深度分页查询（如 from: 10000, size: 10）
• 脚本查询（Script Query）
• 大规模嵌套聚合查询

为了防止这些查询对 Elasticsearch 集群造成影响，我们可以使用 INFINI Gateway 对这些查询进行阻断。

请求上下文

INFINI Gateway 运行环境中有非常多的信息可被利用，而请求上下文就是访问这些信息的入口。如请求来源、请求体信息等，都可使用关键字 _ctx 作为前缀访问相应的上下文信息。

HTTP 请求内置的 _ctx 上下文对象主要包括如下：

更多的上下文信息请访问文档。

context_filter

Context Filter 是 INFINI Gateway 提供的一种在线过滤器，能够根据请求上下文来过滤流量。通过定义一组匹配规则，可以灵活地对流量进行筛选。该过滤器支持多种匹配模式，包括：

• 前缀匹配
• 后缀匹配
• 模糊匹配
• 正则匹配

对于匹配到的请求，可以直接阻断（拒绝）并返回自定义的消息。因此，关键点就是要明确不合理请求的关键字信息。

使用步骤

1. 确定关键字信息：确定特殊查询请求中的关键特征或关键字。
2. 配置匹配规则：在 context_filter 中定义相应的匹配规则，选择合适的匹配模式（如前缀、后缀、模糊或正则匹配）。
3. 阻断请求：一旦匹配到这些关键字，INFINI Gateway 将自动阻断请求并返回指定的消息。

更多详细内容，请参阅相关 文档。

举个例子

阻止 wildcard 查询（模糊匹配查询），我们先看一个 wildcard 查询的样子。

GET yf-test-1shard/_search
{
  "query": {
    "wildcard": {
      "path.keyword": {
        "value": "/a*"
      }
    }
  }
}

上面的查询，会查询 path 字段，所有以 /a 开头的文档。

第一步：我们可确定关键字是 wildcard，为了进一步限制是 wildcard 查询里的情况，我们可将关键字确定为 wildcard":，因为有时候查询 url 里会有 expand_wildcards 字样。

第二步：编辑 INFINI Gateway 默认配置文件，增加 context_filter 匹配规则。

- name: default_flow
  filter:
    - context_filter:
        context: _ctx.request.to_string
        message: "Request blocked. Reason: Forbidden. Please contact the administrator at 010-111111."
        status: 403
        action: deny
        must_not:
          contain:
            - 'wildcard":'

通过上面的修改，我们在 INFINI Gatway 的默认处理流程开头添加了 context_filter 过滤器，阻止查询请求种带关键字 wildcard": 的查询，并返回消息"Request blocked. Reason: Forbidden. Please contact the administrator at 010-111111."

第三步，测试 wildcard 请求能否被阻断。

可以看到，INFINI Gateway 成功阻止了 wildcard 查询，并返回了我们定义的信息。通过此方法，我们可以阻断高消耗类查询被发送到 ES 集群，避免引发集群性能问题。对业务上合理的需求，我们可以进一步沟通，确定合理的方案。

关于极限网关（INFINI Gateway）

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway，可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

开源地址：https://github.com/infinilabs/gateway，如有相关问题或建议，欢迎提交 PR 或 Issue ！

作者：杨帆，极限科技（INFINI Labs）高级解决方案架构师、《老杨玩搜索》栏目 B 站 UP 主，拥有十余年金融行业服务工作经验，熟悉 Linux、数据库、网络等领域。目前主要从事 Easysearch、Elasticsearch 等搜索引擎的技术支持工作，服务国内私有化部署的客户。

有给 ES 系统导入过数据的小伙伴都知道，给一个正在执行查询的 ES 集群导入数据，可能会影响查询的响应时间。如果导入的数据量还比较大，那请将“可能”两个字去掉。这种操作通常被限定在业务低谷期执行，如果一定要立即操作，则必须非常小心控制写入速度，避免影响到业务查询。

INFINI Easysearch 从 1.8.0 版本开始引入了写入限速功能，靠引擎自身能力对写入速度进行限制。不仅听着简单，实际用起来一点也不麻烦，我们赶紧实战一把。

测试环境

INFINI Easyssearch 1.9.0，单节点集群。

测试方法

loadgen 压测 bulk 写入，每个请求写 1000 个文档，每次测试固定写入 500w 个文档。

./loadgen-linux-amd64 -config ../config/write-yf-test.yml -d 3000 -l 5000

请求示例

{"index": {"_index": "yf-test-1shard","_id": "$[[uuid]]"}}
{"ip": "127.0.0.1", "time": "$[[now_utc_lite]]", "method": "GET","path": "/abc", "http_ver": "1.1", "status_code": "200","body_bytes": "3498","agent": "curl","agent_ver": "7.71.1"}

测试基线

单节点不限速写入测试

压测单个索引，3 主，0 副，写入速度 3.8w docs/s

压测单个索引，1 主，0 副，写入速度 2.5w docs/s

同时压测两个索引，写入速度分别是 3w docs/s 和 1.8w docs/s

节点级别限速

基于引擎层实现的限速功能，支持动态开启。比如我想将节点每秒写入的文档数，限制在 10000 个每秒，直接这样设置：

PUT _cluster/settings
{
  "transient": {
    "cluster.throttle.node.write": true
    "cluster.throttle.node.write.max_requests": 10000,
    "cluster.throttle.node.write.action": "retry"
  }
}

压测单个索引，1 主，0 副，写入速度 1w docs/s

压测单个索引，3 主，0 副，写入速度 1w docs/s

由于是限制整个节点的速度，不论索引分片如何，节点的写入上限被限制在了 10000 个文档每秒。节点上的所有分片共享节点的写入限额。

同时压测两个索引，整个节点写入速度还是 10000 个文档每秒。由于我的压测程序对两个索引的写入量是一样的，所以两个索引各占一半。实际上如果两个索引写入压力不一样，就会有高低。

节点级限速适合对节点写入极限比较清楚的条件下，想在节点层面保障集群稳定，不想细分到具体索引的场景。

索引级别限速

索引级的限速可以针对特定索引进行写入限速，避免响其他索引的读写。在之前的不限速测试中，同时写入两个索引的情况下，yf-test-3shard 能达到每秒近 3w docs/s 的写入速度，另一个索引 yf-test-1shard 能达到每秒近 1.8w docs/s 的写入速度。

接下来，我们只对 yf-test-3shard 进行限速。在索引的设置里配置相应的限流阈值：

PUT yf-test-3shard/_settings
{
    "index.throttle.write.max_requests": 2000,
    "index.throttle.write.action": "retry",
    "index.throttle.write.enable": true
}

限速设置在索引设置里查看到。

设置完限速后同时压测两个索引，yf-test-3shard 索引被限制在了 2000 docs/s 的速度，yf-test-1shard 则有更多的资源写入，达到了 2.3w docs/s 的写入速度，比之前不限速的时候稍高。

通过索引级限速功能，我们成功地限制了索引 yf-test-3shard 的写入速度，索引 yf-test-1shard 的写入并未受到影响。

分片级别限速

分片级限流功能，可限定单个分片允许最大写入速度。它不针对哪个索引，而是针对所有分片。比如我想限制每个分片每秒最多写 2000 个文档。

PUT _cluster/settings
{
  "transient": {
    "cluster.throttle.shard.write": true,
    "cluster.throttle.shard.write.max_requests": 2000,
    "cluster.throttle.shard.write.action": "retry"
  }
}

压测单个索引，1 主，0 副

1 个分片，写入速度 2000 个文档每秒。

压测单个索引，3 主，0 副

3 个分片，写入速度 6000 个文档每秒。

不论是哪个索引，全都限定一个分片 2000 的写入速度。我想这种限速比较适合一个集群中有高低配置混搭主机的场景，高配机器性能强悍，磁盘空间也大，分布的分片也多；低配主机性能和磁盘容量都有限，分布的分片数较少。你们说呢？

注意事项

节点级别限流是针对所有 DataNode。

分片级别限流只计算从协调节点分发到数据节点主分片的 bulk 请求。

节点级别和分片级别限流不冲突，可以同时启用。

限流功能不会限制系统索引流量，只针对业务索引。

背景

最近 INFINI Labs 社区有 Easysearch 开发者反馈，其开发环境为 Windows 系统，安装部署 Easysearch 时初始化证书遇到麻烦，如果没有证书就无法开启 Easysearch TLS 传输加密来保护数据的网络传输安全。本文将介绍在 Windows 上创建自签名证书的 7 种不同方法。

使用在线工具 certificatetools.com

在允许生成自签名证书的在线服务中，CertificateTools 是最先进的。只需查看所有可用选项即可：

就这么简单！

使用 Let’s Encrypt

首先，安装 Certbot，这是 Let’s Encrypt 官方推荐的工具，用于自动化获取和续期 SSL/TLS 证书。

1. 安装 Certbot

• 访问 Certbot 下载页面。
• 选择 Windows 系统，下载并安装 Certbot。

2. 获取证书

• 打开 命令提示符 或 PowerShell 以管理员身份运行。
• 输入以下命令获取证书（替换 example.com 为你的域名）：

certbot certonly --standalone --preferred-challenges http -d example.com

• Certbot 会自动通过 HTTP 验证域名并生成证书。证书会存储在：

C:\Certbot\live\example.com\

里面有以下文件：

• cert.pem：证书。
• privkey.pem：私钥。
• fullchain.pem：完整证书链。

3. 导入证书

• 打开 Windows 证书管理器 (mmc)，选择 个人 文件夹。
• 右键点击 个人 文件夹，选择 导入，导入 cert.pem 和 privkey.pem。

4. 验证证书

• 在证书管理器中，确认证书已成功导入并配置。

5. 续期证书

• 使用以下命令手动续期证书：

certbot renew

使用 OpenSSL

OpenSSL 是一个跨平台的工具，适用于各种操作系统，包括 Windows。在 Windows 上，你需要首先安装 OpenSSL。

步骤：

1. 从 OpenSSL 官方网站 下载并安装 OpenSSL。
2. 打开 命令提示符 或 PowerShell，并导航到 OpenSSL 的安装目录。
3. 运行以下命令生成自签名证书：

openssl req -new -x509 -keyout mycert.pem -out mycert.pem -days 365

-new：创建一个新的证书请求。
-x509：生成一个自签名证书。
-keyout 和 -out：指定证书和私钥文件的保存路径。
-days 365：证书有效期为 365 天。

4. 系统会提示你输入一些证书的详细信息，如国家、组织名等。

验证：

检查生成的 mycert.pem 文件是否存在，并通过命令 openssl x509 -in mycert.pem -text 查看证书的内容。

使用 PowerShell

PowerShell 提供了一个简单的命令 New-SelfSignedCertificate 来创建自签名证书。以下是具体的操作步骤：

步骤：

1. 按下 Windows + X，选择 Windows PowerShell (管理员)。
2. 在 PowerShell 窗口中输入以下命令：

New-SelfSignedCertificate -DnsName "example.com" -CertStoreLocation "cert:\LocalMachine\My"

-DnsName "example.com"：指定证书的 DNS 名称，可以更改为你需要的域名或主机名。
-CertStoreLocation "cert:\LocalMachine\My"：将证书存储到本地计算机的证书存储区。

3. 执行后，证书将被创建，并存储在 Windows 证书管理器中。

验证：

1. 打开 运行 (Windows + R)，输入 mmc，点击确定。
2. 在 MMC 中，选择 文件 > 添加/删除管理单元，选择 证书，然后选择 计算机帐户。
3. 查看 个人 文件夹，你将看到刚才创建的证书。

使用 IIS

IIS（Internet Information Services）是一种 Web 服务器软件，可以通过它为你的服务器生成自签名证书。

步骤：

1. 打开 IIS 管理器，选择你的服务器名称。
2. 在主界面中，双击 服务器证书 选项。
3. 在右侧操作面板中，点击 创建自签名证书。
4. 输入证书的名称（如：example.com），然后选择证书的存储位置。
5. 点击确定，证书将被创建并存储在 IIS 中。

验证：

在 服务器证书 部分，你将看到已创建的证书。

使用 MMC 管理工具

Windows 提供了 MMC 管理工具，可以通过图形界面创建自签名证书。

步骤：

1. 按 Windows + R 打开运行窗口，输入 mmc 并按下回车。
2. 在 MMC 中，选择 文件 > 添加/删除管理单元，点击 证书 并选择 计算机帐户。
3. 选择 本地计算机 > 确定。
4. 在左侧的证书树中，右键点击 个人 文件夹，选择 所有任务 > 请求新证书。
5. 跟随向导填写证书的详细信息并选择 自签名证书 选项，完成后证书将被创建。

验证：

在 MMC 中查看证书是否已经生成，并且可以在 个人 文件夹中找到它。

使用 XCA 工具

XCA 是一个开源工具，支持生成和管理证书。它为用户提供了一个图形化界面，适合那些不熟悉命令行操作的用户。

步骤：

1. 从 XCA 官方网站 下载并安装 XCA。
2. 启动 XCA，点击 文件 > 新建数据库 来创建一个新的证书数据库。
3. 在 证书 选项卡中，点击 新建证书。
4. 在证书的设置中，选择 自签名证书，然后填写证书的详细信息。
5. 点击 保存 来生成证书。

验证：

生成的证书可以在 XCA 的 证书 列表中查看，并导出为文件或在需要的地方使用。

总结

在 Windows 上创建自签名证书对于开发者和管理员来说是一项常见任务。自签名证书通常用于测试环境、开发、或者是没有商业证书的情况下使用。本文所述在 Windows 上创建自签名证书的 7 种方法都有详细步骤和验证方式，希望能给你带来帮助。

参考资料

1. 7 ways to create self-signed certificates on Windows

2. 如何使用 XCA 创建证书

Easysearch 的很多用户都有这样的需要，之前是用的 ES，现在要迁移到 Easysearch，但是业务方使用的是 Spring Boot 集成的客户端，问是否能平滑迁移。

Easysearch 是完全兼容 Spring Boot 的，完全不用修改，本指南将探讨如何将 Spring Boot 和 ES 的 high-level 客户端 与 Easysearch 进行集成，涵盖从基础设置到实现 CRUD 操作和测试的所有内容。

服务器设置

首先，需要修改 Easysearch 节点的 easysearch.yml 文件，打开并配置这 2 个配置项：

elasticsearch.api_compatibility: true

#根据客户端版本配置版本号，我这里配置成 7.17.18
elasticsearch.api_compatibility_version: "7.17.18"

项目设置

然后，让我们设置 Maven 依赖。以下是 pom.xml 中的基本配置：

<properties>
    <java.version>11</java.version>
    <spring-data-elasticsearch.version>4.4.18</spring-data-elasticsearch.version>
    <elasticsearch.version>7.17.18</elasticsearch.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.data</groupId>
        <artifactId>spring-data-elasticsearch</artifactId>
        <version>${spring-data-elasticsearch.version}</version>
    </dependency>
    <dependency>
        <groupId>org.elasticsearch</groupId>
        <artifactId>elasticsearch</artifactId>
        <version>${elasticsearch.version}</version>
    </dependency>
    <dependency>
        <groupId>org.elasticsearch.client</groupId>
        <artifactId>elasticsearch-rest-high-level-client</artifactId>
        <version>${elasticsearch.version}</version>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>
</dependencies>

客户端连接配置

完全和连接 Elasticsearch 的方式一样，不用修改：

配置 src/main/resources/application.yml 文件

spring:
  elasticsearch:
    rest:
      uris: https://localhost:9202
      username: admin
      password: xxxxxxxxxxx
    ssl:
      verification-mode: none

连接配置类

@Configuration
public class ElasticsearchConfig extends AbstractElasticsearchConfiguration {
    @Value("${spring.elasticsearch.rest.uris}")
    private String elasticsearchUrl;

    @Value("${spring.elasticsearch.rest.username}")
    private String username;

    @Value("${spring.elasticsearch.rest.password}")
    private String password;

    @Override
    @Bean
    public RestHighLevelClient elasticsearchClient() {
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY,
                new UsernamePasswordCredentials(username, password));

        SSLContext sslContext = SSLContexts.custom()
                .loadTrustMaterial(null, (x509Certificates, s) -> true)
                .build();

        RestClientBuilder builder = RestClient.builder(HttpHost.create(elasticsearchUrl))
                .setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder
                        .setDefaultCredentialsProvider(credentialsProvider)
                        .setSSLContext(sslContext)
                        .setSSLHostnameVerifier(NoopHostnameVerifier.INSTANCE));

        return new RestHighLevelClient(builder);
    }
}

领域模型

使用 Spring 的 Elasticsearch 注解定义领域模型：

@Data
@Document(indexName = "products")
public class Product {
    @Id
    private String id;

    @Field(type = FieldType.Text, name = "name")
    private String name;

    @Field(type = FieldType.Double, name = "price")
    private Double price;
}

仓库层

创建继承 ElasticsearchRepository 的仓库接口：

@Repository
@EnableElasticsearchRepositories
public interface ProductRepository extends ElasticsearchRepository<Product, String> {
}

服务层

实现服务层来处理业务逻辑：

@Service
public class ProductService {
    private final ProductRepository productRepository;

    @Autowired
    public ProductService(ProductRepository productRepository) {
        this.productRepository = productRepository;
    }

    public Product saveProduct(Product product) {
        return productRepository.save(product);
    }

    public Product findProductById(String id) {
        return productRepository.findById(id).orElse(null);
    }
}

测试

编写集成测试类：

@SpringBootTest
public class ProductServiceIntegrationTest {
    @Autowired
    private ElasticsearchOperations elasticsearchOperations;

    @Autowired
    private ProductService productService;

    private static final String INDEX_NAME = "products";

    @BeforeEach
    public void setUp() {
        IndexOperations indexOperations = elasticsearchOperations.indexOps(IndexCoordinates.of(INDEX_NAME));
        if (indexOperations.exists()) {
            indexOperations.delete();
        }

        // 定义 mapping
        Document mapping = Document.create()
                .append("properties", Document.create()
                        .append("name", Document.create()
                                .append("type", "text")
                                .append("analyzer", "standard"))
                        .append("price", Document.create()
                                .append("type", "double")));

        // 创建索引并应用 mapping
        indexOperations.create(Collections.EMPTY_MAP, mapping);
    }

    @Test
    public void testSaveAndFindProduct() {
         List<Product> products = Arrays.asList(
                new Product("Test Product 1", 99.99),
                new Product("Test Product 2", 199.99),
                new Product("Test Product 3", 299.99)
        );

        List<IndexQuery> queries = products.stream()
            .map(product -> new IndexQueryBuilder()
                .withObject(product)
                .withIndex(INDEX_NAME)
                .build())
            .collect(Collectors.toList());

        List<IndexedObjectInformation> indexedInfos = elasticsearchOperations.bulkIndex(
            queries,
            IndexCoordinates.of(INDEX_NAME)
        );

        // 验证结果
        List<String> ids = indexedInfos.stream()
            .map(IndexedObjectInformation::getId)
            .collect(Collectors.toList());

        assertFalse(ids.isEmpty());
        assertEquals(products.size(), ids.size());
    }
}

结论

本指南展示了 Easysearch 与 Elasticsearch 的高度兼容性：

1. 配置方式相同，仅需启用 Easysearch 的 API 兼容模式。
2. 可直接使用现有 Elasticsearch 客户端。
3. Maven 依赖无需更改。
4. API、注解和仓库接口完全兼容。
5. 现有测试代码可直接应用。

这种兼容性使得从 Elasticsearch 迁移到 Easysearch 成为一个简单、低风险的过程。Spring Boot 项目可以几乎无缝地切换到 Easysearch，同时获得其性能和资源利用方面的优势。

关于 Easysearch

INFINI Easysearch 是一个分布式的搜索型数据库，实现非结构化数据检索、全文检索、向量检索、地理位置信息查询、组合索引查询、多语种支持、聚合分析等。Easysearch 可以完美替代 Elasticsearch，同时添加和完善多项企业级功能。Easysearch 助您拥有简洁、高效、易用的搜索体验。

官网文档：https://infinilabs.cn/docs/latest/easysearch

作者：张磊，极限科技（INFINI Labs）搜索引擎研发负责人，对 Elasticsearch 和 Lucene 源码比较熟悉，目前主要负责公司的 Easysearch 产品的研发以及客户服务工作。
原文：https://infinilabs.cn/blog/2024/use-spring-boot-for-easysearch-connection/

在日常使用 Easysearch 中，难免会遇到集群密码需要重置的情况（如密码遗失、安全审计要求等）。

通过查看 Easysearch 用户接口文档，创建用户使用如下接口：

PUT _security/user/<username>
{
  "password": "adminpass",
  "roles": ["maintenance_staff", "weapons"],
  "external_roles": ["captains", "starfleet"],
  "attributes": {
    "attribute1": "value1",
    "attribute2": "value2"
  }
}

同样可以通过该接口对用户进行密码重置：

PUT _security/user/<username>
{
  "password": "adminpass111",
  "roles": ["maintenance_staff", "weapons"],
  "external_roles": ["captains", "starfleet"],
  "attributes": {
    "attribute1": "value1",
    "attribute2": "value2"
}

API 接口创建的用户

创建一个测试用户 test，并进行访问验证。

通过接口重置用户 test 密码，并进行访问验证。

配置文件创建的用户

在配置文件 user.yml 中添加测试用户（test1、test2）配置：

## Test users
test1:
  hash: "$2y$12$ZNfKKCeeRQXOWX27W50tbu0Tq4NT4ADdCQOBoZzokI1zR8ZEUWm4W" # test1
  reserved: true
  roles:
    - "readall_and_monitor"

test2:
  hash: "$2y$12$m4/eSiDlzRII87vNeKwzteEHGpgpbdMr5tRvOfve/xIbEYdC4bU7a" # test2
  reserved: false
  roles:
    - "readall_and_monitor"

其中 hash 字段是将用户密码哈希出来的值，可通过 bin/hash_password.sh -p "<明文密码>" 生成。

进行用户访问验证。

对 test1、test2 用户进行密码重置。

用户 test1 重置失败，用户 test2 重置成功。用户 test1 重置失败是因为配置了 reserved: true，将用户 test1 指定为内置用户，使用用户 admin 会因为权限无法进行密码重置，那就需要使用有更高权限的管理证书进行密码重置。默认在 Easysearch 集群执行初始化脚本 bin/initialize.sh 时，会在 config 目录下生成证书文件，其中 admin.crt、admin.key 为管理证书。

test1 用户密码重置成功，进行访问验证。

关于 Easysearch

INFINI Easysearch 是一个分布式的搜索型数据库，实现非结构化数据检索、全文检索、向量检索、地理位置信息查询、组合索引查询、多语种支持、聚合分析等。Easysearch 可以完美替代 Elasticsearch，同时添加和完善多项企业级功能。Easysearch 助您拥有简洁、高效、易用的搜索体验。

官网文档：https://infinilabs.cn/docs/latest/easysearch

在日志场景，还是有很多小伙伴在使用 Filebeat 采集日志的。今天我来实战下使用 Filebeat 8 连接 Easysearch 。本次使用 Easysearch-1.9.0 版本和 Filebeat-8.17.0 版本做演示，也适用 Filebeat-oss-8.17.0 版本。

Easysearch 不开启兼容参数的情况

Easysearch 默认情况下未开启 Elastic 兼容功能。此时直接用 Filebeat 去连接 Easysearch 会失败，报错 "could not connect to a compatible version of Elasticsearch" ，直译过来就是“无法连接到兼容的 Elasticsearch 版本”。

Easysearch 开启 elasticsearch.api_compatibility 参数

在 Easysearch-1.9.0 中关于 Elasticsearch 的兼容性参数有两个，我们先开第一个。

elasticsearch.api_compatibility: true
#elasticsearch.api_compatibility_version: "8.9.0"

启动 Filebeat 连接。

这次 Filebeat 检测到了一个兼容的版本 7.10.2，并导入了一个叫 filebeat 的索引生命周期管理策略，但最终因为无法导入模板而失败了。

索引生命周期查看。

关闭 Filebeat 的导入模板功能尝试连接，修改配置 filebeat.yml 关闭模板导入。

setup.template.enabled: false

启动 Filebeat 连接 Easysearch。

这次 Filebeat 成功连接上了 Easysearch，并成功将采集数据写入。

Easysearch 同时开启两个兼容性参数

在 Easysearch-1.9.0 中关于 Elasticsearch 的兼容性参数有两个，这次我们二个同时打开。

elasticsearch.api_compatibility: true
elasticsearch.api_compatibility_version: "8.9.0"

清理测试数据后，启动 Filebeat 连接。

跟打开第一个兼容参数很像，只是 Filebeat 检测到的版本变成了 8.9.0。

我们关闭 Filebeat 的模板导入功能，再次连接——成功。

数据也成功写入。

注意事项

Easysearch 的第一个兼容参数 elasticsearch.api_compatibility 必须开启，否则 Filebeat 检测到的还是 1.9.0 版本无法正常连接。

如果 Easysearch 使用的是 1.9.0 之前的版本，可以使用网关代理 Easysearch 的方法。

总结

Easysearch-1.9.0 必须开启第一个兼容参数，此时检测到的版本是 7.10.2。也可同时开启两个兼容性参数，此时检测到的版本是 8.9.0。

Filebeat 必须关闭模板导入功能。

好了，还有其他问题的小伙伴，欢迎加我微信沟通。

背景

很多小伙伴都在使用 Logstash ，随着各家安全扫描、安全策略的加固，不少小伙伴已经开始使用 Logstash 8 了。在使用 Logstash 8 连接 Easysearch 的时候可能会遇到问题，比如下图。

提示连接的不是兼容版本的 Elasticsearch 。

解决方法有两种

两种方法对 Logstash 和 Logstash-oss 两种版本都适用。

一、用 INFINI Gateway 代理 Easysearch

使用此方法， Logstash 连接的是 INFINI Gateway ，由 INFINI Gateway “通过” Logstash 8 的检查。

优点是对 Easysearch 版本无要求。

默认的网关配置要稍作修改。

1. 修改 router.rules 部分

router:
  - name: my_router
    default_flow: default_flow
    tracing_flow: logging_flow
    rules:
      - method:
          - "GET"
        pattern:
          - "/"
        flow:
          - overwrite_flow
      - method:
          - "GET"
        pattern:
          - "/_license"
        flow:
          - overwrite_license_flow
      - method:
          - "*"
        pattern:
          - "/_bulk"
          - "/{any_index}/_bulk"
        flow:
          - async_bulk_flow

2. flow 下增加 overwrite_flow 和 overwrite_license_flow

 - name: overwrite_flow
    filter:
      - elasticsearch:
          elasticsearch: prod
          max_connection_per_node: 1000
      - set_context:
          context:
            _ctx.response.body_json.version.number: '"8.14.3"'
            _ctx.response.body_json.version.build_flavor: '"default"'
            _ctx.response.body_json.tagline: '"You Know, for Search"'
            _ctx.response.body_json.minimum_wire_compatibility_version: '"7.17.0"'
            _ctx.response.body_json.minimum_index_compatibility_version: '"7.0.0"'
      - set_response_header:
          headers:
            - X-elastic-product -> Elasticsearch
  - name: overwrite_license_flow
    filter:
      - echo:
          message: '{"license": "ok"}'
      - set_context:
          context:
            _ctx.response.code: 200
            _ctx.response.body_json.license.status: '"active"'
            _ctx.response.body_json.license.type: '"basic"'
            _ctx.response.body_json.license.issuer: '"elasticsearch"'
            _ctx.response.body_json.license.issue_date: '"2024-09-12T13:10:55.848Z"'
            _ctx.response.body_json.license.start_date_in_millis: "-1"
      - set_response_header:
          headers:
            - X-elastic-product -> Elasticsearch

记住 Logstash 要填写 INFINI Gateway 的地址和端口，并关闭自动导入模板功能。

连接成功后，如下图。

二、使用 Easysearch 1.9.0 及以上版本

Easysearch 1.9.0 已经增加了适配 Logstash8.x 的请求 header，因此使用 Easysearch 1.9.0 的小伙伴，只需打开对应的设置就可以了。

在 easysearch.yml 中，打开下面的 elastic 的兼容参数（默认已注释）：

elasticsearch.api_compatibility: true
elasticsearch.api_compatibility_version: "8.9.0"

注意 Logstash 配置中关闭自动导入模板功能。

output {
  elasticsearch {
   hosts => ["http://127.0.0.1:9200"]
   index => "logstash8"
   manage_template => false
}
}

启动 Logstash 连接到 easysearch 后， Logstash 日志会输出检测到版本 8.9.0 。

最新 Easysearch 下载地址：下载

好了，还有其他问题的小伙伴，欢迎加我微信沟通。

压测环境

虚拟机配置

使用阿里云上规格：ecs.u1-c1m4.4xlarge，PL2: 单盘 IOPS 性能上限 10 万 (适用的云盘容量范围：461GiB - 64TiB)

Easysearch 配置

7 节点集群，版本：1.9.0

Elasticsearch 配置

7 节点集群，版本：7.10.2

监控集群配置

单节点 Easysearch 集群，版本：1.9.0

压测 loadgen 配置

loadgen 版本：1.25.0

4 台压 Easysearch，4 台压 Elasticsearch。

压测索引 Mapping

PUT nginx
{
  "mappings": {
    "properties": {
      "method": {
        "type": "keyword"
      },
      "bandwidth": {
        "type": "integer"
      },
      "service_name": {
        "type": "keyword"
      },
      "ip": {
        "type": "ip"
      },
      "memory_usage": {
        "type": "integer"
      },
      "upstream_time": {
        "type": "float"
      },
      "url": {
        "type": "keyword"
      },
      "response_size": {
        "type": "integer"
      },
      "request_time": {
        "type": "float"
      },
      "request_body_size": {
        "type": "integer"
      },
      "error_code": {
        "type": "keyword"
      },
      "metrics": {
        "properties": {
          "queue_size": {
            "type": "integer"
          },
          "memory_usage": {
            "type": "integer"
          },
          "thread_count": {
            "type": "integer"
          },
          "cpu_usage": {
            "type": "integer"
          },
          "active_connections": {
            "type": "integer"
          }
        }
      },
      "cpu_usage": {
        "type": "integer"
      },
      "user_agent": {
        "type": "keyword"
      },
      "connections": {
        "type": "integer"
      },
      "timestamp": {
        "type": "date",
        "format": "yyyy-MM-dd'T'HH:mm:ss.SSS"
      },
      "status": {
        "type": "integer"
      }
    }
  },
  "settings": {
    "number_of_shards": 7,
    "number_of_replicas": 0,
    "refresh_interval": "30s"
  }
}

压测方法

每 4 个 loadgen 使用批量写入接口 bulk 轮询压测同一集群的 7 个节点，每个请求写入 10000 个文档。

具体请求如下：

requests:
  - request: #prepare some docs
      method: POST
      runtime_variables:
#        batch_no: uuid
      runtime_body_line_variables:
#        routing_no: uuid
#      url: $[[env.ES_ENDPOINT]]/_bulk
      url: $[[ip]]/_bulk
      body_repeat_times: 10000
      basic_auth:
       username: "$[[env.ES_USERNAME]]"
       password: "$[[env.ES_PASSWORD]]"
      body: |
        {"index": {"_index": "nginx", "_type": "_doc", "_id": "$[[uuid]]"}}
        $[[message]]

压测数据样本

{"method":"DELETE","bandwidth":1955,"service_name":"cart-service","ip":"120.204.26.240","memory_usage":1463,"upstream_time":"1.418","url":"/health","response_size":421,"request_time":"0.503","request_body_size":1737,"error_code":"SYSTEM_ERROR","metrics":{"queue_size":769,"memory_usage":1183,"thread_count":65,"cpu_usage":68,"active_connections":837},"cpu_usage":70,"user_agent":"Mozilla/5.0 (iPad; CPU OS 14_6 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.1.1","connections":54,"timestamp":"2024-11-16T14:25:21.423","status":500}
{"method":"OPTIONS","bandwidth":10761,"service_name":"product-service","ip":"223.99.83.60","memory_usage":567,"upstream_time":"0.907","url":"/static/js/app.js","response_size":679,"request_time":"1.287","request_body_size":1233,"error_code":"NOT_FOUND","metrics":{"queue_size":565,"memory_usage":1440,"thread_count":148,"cpu_usage":39,"active_connections":1591},"cpu_usage":87,"user_agent":"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.1.1","connections":354,"timestamp":"2024-11-16T05:37:28.423","status":502}
{"method":"HEAD","bandwidth":10257,"service_name":"recommendation-service","ip":"183.60.242.143","memory_usage":1244,"upstream_time":"0.194","url":"/api/v1/recommendations","response_size":427,"request_time":"1.449","request_body_size":1536,"error_code":"UNAUTHORIZED","metrics":{"queue_size":848,"memory_usage":866,"thread_count":86,"cpu_usage":29,"active_connections":3846},"cpu_usage":71,"user_agent":"Mozilla/5.0 (compatible; Googlebot/2.1; +http://www.google.com/bot.html)","connections":500,"timestamp":"2024-11-16T15:14:30.424","status":403}

压测索引 1 主分片 0 副本

Elastic 吞吐

Elastic 线程及队列

资源消耗

Easysearch 吞吐

Easysearch 线程及队列

资源消耗

对比

压测索引 1 主分片 1 副本

Elastic 吞吐

Elastic 线程及队列

资源消耗

Easysearch 吞吐

Easysearch 线程及队列

资源消耗

对比

压测索引 7 主分片

Elastic 吞吐

Elastic 线程及队列

资源消耗

网络

单节点平均接收 26MB/s，对应带宽：1456 Mb/s

5 千万文档，总存储 105 GB，单节点 15 GB

Easysearch 吞吐

Easysearch 线程及队列

资源消耗

对比

总结

通过对不同场景的压测结果进行对比分析，得出以下结论：

• Easysearch 相比 Elasticsearch 的索引性能显著提升
  Easysearch 集群的吞吐性能提升了 40% - 70%，且随着分片数量的增加，性能提升效果更为显著。
• Easysearch 相比 Elasticsearch 的磁盘压缩效率大幅提高
  Easysearch 集群的磁盘压缩效率提升了 2.5 - 3 倍，并且随着数据量的增加，压缩效果愈发明显。

此测试结果表明，Easysearch 在日志处理场景中具有更高的性能与存储效率优势，尤其适用于大规模分片与海量数据的使用场景。

如有任何问题，请随时联系我，期待与您交流！

背景

在现代数据驱动的世界中，时序数据的处理变得越来越重要。无论是监控系统、日志分析，还是物联网设备的数据收集，时序数据都占据了大量的存储空间。随着时间的推移，这些数据的存储成本和管理复杂度也在不断增加。

为了解决这一问题，Rollup 技术应运而生。本文将带你深入了解 Rollup 的概念、优势以及如何在 Easysearch 中使用 Rollup 来优化时序数据的存储和查询。

什么是 Rollup？

Rollup 是一种数据聚合技术，主要用于处理时序数据。它通过将细粒度的原始数据汇总为粗粒度的数据，从而减少存储空间并提高查询效率。简单来说，Rollup 可以将大量的详细数据压缩为更小的、更具代表性的数据集，同时保留关键的业务指标。

Rollup 的优势

1. 降低存储成本：通过汇总数据，Rollup 可以显著减少历史数据的存储空间，从而降低存储成本。
2. 提高查询性能：汇总后的数据量更小，查询速度更快，尤其是在处理大规模历史数据时，性能提升尤为明显。
3. 无缝集成：Easysearch 的 Rollup 功能支持直接查询原始索引，业务代码无需修改，用户无感知。
4. 自动化管理：Rollup 任务可以自动滚动生成新的索引，无需手动干预，简化了数据管理流程。

Rollup 的使用场景

Rollup 特别适用于以下场景：

• 监控系统：监控系统产生的指标数据通常非常详细，但随着时间推移，这些数据的价值逐渐降低。通过 Rollup，可以将这些数据汇总为更粗粒度的指标，保留关键信息的同时减少存储压力。
• 日志分析：日志数据通常包含大量的细节信息，但随着时间的推移，这些细节信息的重要性逐渐降低。Rollup 可以帮助你将日志数据汇总为更高层次的统计信息，便于长期存储和分析。
• 物联网数据：物联网设备产生的数据量巨大，且大部分数据在短期内具有较高的分析价值。通过 Rollup，可以将这些数据汇总为更粗粒度的形式，便于长期存储和查询。

使用 Rollup 的先决条件

在使用 Rollup 之前，需要满足以下条件：

安装索引生命周期管理插件：Rollup 功能是索引生命周期管理插件的一部分，因此必须安装该插件。

源索引必须包含 date 类型字段：Rollup 依赖于时间字段来进行数据聚合，因此源索引必须包含一个 date 类型的字段。

Easysearch 中的 Rollup 功能

Easysearch 提供了强大的 Rollup 功能，支持多种聚合类型，并且可以自动滚动生成新的索引。下面我们将详细介绍如何在 Easysearch 中使用 Rollup。

支持的聚合类型

Easysearch 的 Rollup 功能支持以下聚合类型：

• 数值类型字段：avg、sum、max、min、value_count、percentiles
• keyword 类型字段：terms 聚合
• date 类型字段：date_histogram 和 date_range 聚合

Rollup 的核心参数详解

在配置 Rollup 任务时，以下几个参数至关重要：

1. metrics 参数：定义需要聚合的数值字段

metrics 参数用于指定哪些数值字段需要进行聚合计算。Rollup 任务会对这些字段进行指定的聚合操作（如 avg、sum、max、min、value_count、percentiles），并将结果存储到目标索引中。

示例：

"metrics": [
  "payload.elasticsearch.index_stats.*"
]

在这个例子中，metrics 指定了对 payload.elasticsearch.index_stats 下的所有字段进行聚合。

2. attributes 参数：保留原始数据的非聚合字段

attributes 参数用于指定哪些字段需要原封不动地保留在 Rollup 结果中。这些字段不会被聚合，而是直接复制到目标索引中。

示例：

"attributes": [
  "agent.*",
  "metadata.*"
]

在这个例子中，attributes 指定了所有以 agent. 和 metadata. 开头的字段都会被保留到 Rollup 结果中。

3. exclude 参数：排除不需要处理的字段

exclude 参数用于排除某些字段，使其不参与 Rollup 任务。这些字段既不会被聚合，也不会被保留到目标索引中。

示例：

"exclude": ["payload.elasticsearch.index_stats.routing.*"]

在这个例子中，exclude 指定了 payload.elasticsearch.index_stats.routing 下的所有字段都不会被处理。

4. filter 参数：过滤源数据

filter 参数用于过滤源数据，只有符合过滤条件的文档才会被 Rollup 任务处理。这个参数可以帮助你减少需要处理的数据量，从而提高 Rollup 任务的效率。

示例：

"filter": {
  "metadata.name": "index_stats"
}

在这个例子中，filter 指定了只有 metadata.name 字段值为 index_stats 的文档才会被 Rollup 任务处理。

5. identity 参数：定义分组字段

identity 参数用于指定哪些字段作为分组条件。Rollup 任务会根据这些字段的值对数据进行分组，然后在每个分组内进行聚合计算。

功能：

• 定义数据分组的字段。
• 支持多个字段的组合，用于创建唯一的分组键。
• 常用于标识数据的来源或类别。

示例：

"identity": [
  "metadata.labels.cluster_id",
  "metadata.labels.index_id",
  "metadata.labels.index_name",
  "metadata.category",
  "payload.elasticsearch.index_stats.index_info.health"
]

在这个例子中，identity 指定了多个字段作为分组条件。Rollup 任务会根据这些字段的值对数据进行分组，然后在每个分组内进行聚合计算。

6. interval 参数：定义时间聚合间隔

interval 参数用于指定时间聚合的间隔。Rollup 任务会根据这个间隔将数据按时间分桶，然后在每个时间桶内进行聚合计算。

功能：

• 定义时间聚合的粒度。
• 支持多种时间单位，如 1m（1 分钟）、1h（1 小时）、1d（1 天）等。
• 常用于按时间维度汇总数据。

示例：

"interval": "1m"

在这个例子中，interval 指定了时间聚合的间隔为 1 分钟。Rollup 任务会每分钟对数据进行一次聚合。

1.10 版本的新特性

从 Easysearch 1.10.0 版本开始，Rollup 功能引入了一些新特性，进一步增强了其灵活性和易用性。

1. 支持 date_range 聚合

在 1.10.0 版本中，Rollup 增加了对 原始索引使用 date_range 聚合的支持。这意味着你可以在原始索引根据日期范围对数据进行聚合，而不仅仅是固定的时间间隔。

示例：

"date_range": {
  "field": "@timestamp",
  "ranges": [
    { "from": "now-1d/d", "to": "now" },
    { "from": "now-7d/d", "to": "now-1d/d" }
  ]
}

2. 通配符方式批量启动/停止 Rollup Job

在 1.10.0 版本中，你可以使用通配符批量启动或停止 Rollup 任务。这大大简化了任务管理的操作。

示例：

POST _rollup/jobs/rollup*/_start
POST _rollup/jobs/rollup*/_stop

3. 设置 Rollup 索引自动滚动的条数

你可以通过设置 rollup.max_docs 参数，控制 Rollup 索引自动滚动的条数。当索引中的文档数量达到设定值时，系统会自动创建一个新的 Rollup 索引。

示例：

PUT /_cluster/settings
{
    "transient": {
      "rollup.max_docs": 10000000
    }
}

4. 新增 ROLLUP_SEARCH_MAX_COUNT 配置

在 1.10.0 版本中，新增了 ROLLUP_SEARCH_MAX_COUNT 配置项，用于控制 Rollup 在运行 Job 时收集历史数据的最大并发分片请求数。这个配置项可以帮助你优化 Rollup 任务的性能，并避免集群资源过载。

功能：

• 控制并发请求数：限制 Rollup 任务在执行搜索请求时的最大并发分片请求数。
• 动态调整：支持在集群运行时动态调整，无需重启集群。
• 默认值：2，即默认情况下，Rollup 任务最多会同时发送 2 个并发分片请求。

示例：

PUT /_cluster/settings
{
    "transient": {
      "rollup.search.max_count": 2
    }
}

在这个例子中，ROLLUP_SEARCH_MAX_COUNT 被设置为 2，表示 Rollup 任务在执行搜索请求时，最多会同时发送 2 个并发分片请求。

配置建议：

• 小规模集群：建议设置为较小的值（如 2），以避免资源竞争。
• 大规模集群：可以适当增加该值（如 4），以提高并发性能。
• 动态调整：根据集群负载情况动态调整该值，以优化性能和资源利用率。

创建 Rollup 任务的完整示例

以下是一个完整的 Rollup 任务配置示例，展示了 metrics、attributes、exclude 和 filter 参数的综合使用：

PUT _rollup/jobs/rollup1
{
  "rollup": {
    "source_index": ".infini_metrics",
    "target_index": "rollup1_{{ctx.source_index}}",
    "timestamp": "timestamp",
    "continuous": true,
    "page_size": 1000,
    "cron": "*/10 1-23 * * *",
    "timezone": "UTC+8",
    "stats": [
      {
        "max": {}
      },
      {
        "value_count": {}
      }
    ],
    "interval": "1m",
    "identity": [
      "metadata.labels.cluster_id",
      "metadata.labels.index_id",
      "metadata.labels.index_name",
      "metadata.category",
      "payload.elasticsearch.index_stats.index_info.health"
    ],
    "attributes": [
      "agent.*",
      "metadata.*"
    ],
    "metrics": [
      "payload.elasticsearch.index_stats.*"
    ],
    "exclude": ["payload.elasticsearch.index_stats.routing.*"],
    "filter": {
      "metadata.name": "index_stats"
    }
  }
}

如何使用 Rollup 索引

从 1.10.0 版本开始，索引生命周期插件不再默认启用 rollup 搜索功能，如果想使用搜索 rollup 搜索功能，需要设置

PUT /_cluster/settings
{
    "transient": {
      "rollup.search.enabled": true
    }
}

无需特意搜索 rollup 索引，只需使用标准的 _search API 对原始目标索引进行搜索。需要注意的是，查询时必须符合目标索引的约束条件。

以下是一个使用 Rollup 索引的示例：

GET target-test/_search
{
  "size": 0,
  "aggs": {
    "a": {
      "date_histogram": {
        "field": "@timestamp",
        "fixed_interval": "1h"
      }
    },
    "total_passenger_count": {
      "sum": {
        "field": "passenger_count"
      }
    }
  }
}

总结

Rollup 是处理时序数据的强大工具，能够有效降低存储成本并提高查询性能。Easysearch 的 Rollup 功能不仅支持多种聚合类型，还提供了自动滚动索引、无缝查询等特性，极大地简化了时序数据的管理和分析流程。通过合理配置 metrics、attributes、exclude 和 filter 参数，你可以灵活地控制 Rollup 任务的行为，从而高效地处理时序数据。

如果你正在处理大量的时序数据，不妨尝试使用 Rollup 来优化你的数据存储和查询。通过本文的介绍，相信你已经对 Rollup 有了深入的了解。赶快动手试试吧，体验 Rollup 带来的高效与便捷！

更详细的使用文档可在 官网 查看

关于 Easysearch

INFINI Easysearch 是一个分布式的搜索型数据库，实现非结构化数据检索、全文检索、向量检索、地理位置信息查询、组合索引查询、多语种支持、聚合分析等。Easysearch 可以完美替代 Elasticsearch，同时添加和完善多项企业级功能。Easysearch 助您拥有简洁、高效、易用的搜索体验。

官网文档：https://infinilabs.cn/docs/latest/easysearch

作者：张磊，极限科技（INFINI Labs）搜索引擎研发负责人，对 Elasticsearch 和 Lucene 源码比较熟悉，目前主要负责公司的 Easysearch 产品的研发以及客户服务工作。

从 Elasticsearch 迁移到 Easysearch 需要考虑多个方面，这取决于当前使用的 Elasticsearch 版本、能容忍的停机时间、应用需求等。在此背景下，我们梳理了一下通用的升级指引，方便大家进行迁移工作。

迁移路径

之前有同事做过相关测试，详情请移步这里。

快照恢复迁移

1. 部署新的 Easysearch 集群，如果有使用插件（如 IK），也一并安装。
2. 将备份仓库注册到 Easysearch 集群。
3. 在 Easysearch 中设置需要使用的用户名和密码信息。
4. 原 Elasticsearch 集群进行快照备份。
5. 在 Easysearch 集群中进行备份还原。
6. 另部署一套应用连接 Easysearch 集群，进行数据、功能验证。
7. 停止应用写入新的数据到 Elasticsearch。
8. 原 Elasticsearch 集群进行快照备份。
9. 在 Easysearch 集群中进行备份还原。
10. 再次使用应用验证数据、功能正常。
11. 切换，老应用下线使用新应用或者老应用修改地址连接 Easysearch 集群。

INFINI Console 迁移

1. 部署新的 Easysearch 集群及其插件（如 IK）。
2. 部署 INFINI Console、Gateway 程序。
3. 将 Elasticsearch 和 Easysearch 注册到 INFINI Console 中。
4. 在 Easysearch 中设置需要使用的用户名和密码信息。
5. 建立数据迁移任务，对业务索引进行迁移，建议启用压缩功能。
6. 另部署一套应用连接 Easysearch 集群，进行数据、功能验证。
7. 停止应用写入新的数据到 Elasticsearch。
8. 再次建立数据迁移任务，设置条件，只迁移增量数据。
9. 再次使用应用验证数据、功能正常。
10. 切换，老应用下线使用新应用或者老应用修改地址连接 Easysearch 集群。

客户端调整

如果要继续使用 Java High Level REST Client，建议将版本调整到 7.10.2 。当然更建议的是使用 Easysearch 的客户端，更轻更快，构建查询，跟搭积木一样简单。

开源事业

极限科技（INFINI Labs） 一直致力于为开发者和企业提供优质的开源工具，提升整个技术生态的活力。除了维护国内最流行的分词器 analysis-ik 和 analysis-pinyin ，也在不断推动更多高质量开源产品的诞生。

在极限科技成立三周年之际，公司宣布以下产品和工具已全面开源：

• INFINI Framework
• INFINI Gateway
• INFINI Console
• INFINI Agent
• INFINI Loadgen
• INFINI Coco AI

以上开源软件都可以在 Github 上面找到：https://github.com/infinilabs

欢迎大家一起参与到开源工具的维护、贡献当中来，别忘了 Star🌟 支持一下！！！

如果您对迁移过程有任何疑问，欢迎与我讨论。

可搜索快照功能改变了我们对备份数据的查询方式。以往要查询备份数据时，要先找到备份数据所在的快照，然后在一个合适的环境中恢复快照，最后再发起请求查询数据。这个处理路径很长，而且很消耗时间。可搜索快照功能将大大简化该处理路径，节约时间。

角色设置

相信你对节点角色的概念已经有所熟悉。要启用可搜索快照功能，Easysearch 集群中必须至少有一个节点拥有 search 角色。参考设置如下。

node.roles: ["search"]
node.search.cache.size: 500mb

• node.roles: 指定节点角色，只有 search 角色的节点才能去搜索快照中的数据。
• node.search.cache.size: 执行快照搜索时，数据缓存大小。

混合角色设置，参考如下。

node.roles: ["master","data","search","ingest"]
node.search.cache.size: 500mb

创建快照

可搜索快照功能使用普通快照作为基础，创建快照命令不变。比如我创建且备份个 infini 索引。

# 创建 infini 索引
POST infini/_doc
{
  "test":"Searchable snapshots"
}

# 创建快照备份 infini 索引
PUT _snapshot/my-fs-repository/1
{
  "indices": "infini",
  "include_global_state": false
}

创建快照索引

可搜索快照功能的核心是搜索快照中的索引，这一步是通过快照索引实现的。为了和集群中的普通索引区别开来，我们将实际存储在快照中的索引称为快照索引。通过使用 Easysearch 的 _restore API 并指定 remote_snapshot 存储类型来创建快照索引。

创建快照索引时，注意名称不能与当前索引名称重复。通常我们备份完索引后，可删除索引释放节点磁盘空间，创建快照索引时默认使用原来的名称。

# 删除 infini 索引释放磁盘空间
DELETE infini

# 创建快照索引,使用原索引名称
POST /_snapshot/my-fs-repository/1/_restore
{
  "indices": "infini",
  "include_global_state": false,
  "include_aliases": false,
  "storage_type": "remote_snapshot"
}

创建快照索引的命令和还原快照的命令非常相似，关键在于 storage_type 参数指定 remote_snapshot 存储类型。

如果要将快照中的全部索引都创建快照索引，可省略 indices 参数。

如果想在创建快照索引时指定不同的名字，参考下面的命令。

POST /_snapshot/my-fs-repository/1/_restore
{
  "indices": "infini",
  "include_global_state": false,
  "include_aliases": false,
  "storage_type": "remote_snapshot",
  "rename_pattern": "(infini)",
  "rename_replacement": "snapshot-$1"
}

• rename_pattern: 使用此选项指定索引匹配的正则表达式。使用捕获组重用索引名称的部分。
• rename_replacement: 使用 $0 包括整个匹配索引名称，使用 $1 包括第一个捕获组的内容，等等。

上述命令创建出来的快照索引名称是 snapshot-infini 。

经过上面一系列的操作，我已经拥有了两个快照索引。

搜索快照索引

我们通过搜索快照索引达到搜索快照数据的目的，令人开心的是搜索快照索引和搜索普通索引的语法完全一样。😀

常见问题

如何区分普通索引和快照索引呢？

我们可以通过索引的 settings 信息区分，快照索引的 settings 信息中有 store.type: remote_snapshot 信息，普通索引没有此信息。

快照索引能写入数据吗？

快照索引无法写入，数据仍然保持在快照格式中存储在存储库中，因此可搜索快照索引本质上是只读的。 任何尝试写入可搜索快照索引的操作都会导致错误。

快照索引不想要了怎么办？

直接删除，需要时再执行创建快照索引流程。此外快照在创建快照索引后，无法直接删除快照，要先删除快照索引。

如果您对上述内容有任何疑问，欢迎与我讨论。