背景

在处理时序数据时，Rollup 功能通过数据聚合显著降低存储成本，并提升查询性能。Easysearch 与 OpenSearch 均提供了 Rollup 能力，但在多个关键维度上，Easysearch Rollup 展现出更优的表现。本文将从查询体验、索引管理、聚合能力、性能优化和任务管理五个方面，分析 Easysearch Rollup 相较于 OpenSearch Rollup 的优势。

Easysearch vs OpenSearch

1. 查询体验：标准接口与无缝集成

Easysearch Rollup 支持通过标准的 _search API 查询原始索引，系统自动融合 Rollup 数据，用户无需变更现有代码或使用专用查询端点。相比之下，OpenSearch Rollup 虽然使用标准查询语法，但需要用户显式指定 Rollup 索引，无法自动结合原始数据，在需要同时访问原始与聚合数据的场景下显得更为繁琐。

• 差异：Easysearch 支持自动融合原始与 Rollup 数据，OpenSearch 需手动指定索引。
• 影响：Easysearch 显著降低了查询逻辑的复杂性和开发维护成本。

2. 索引管理：自动化与扩展能力

Easysearch Rollup 提供自动索引滚动功能，可通过 rollup.index_max_docs. 配置项为不同的目标 Rollup 索引设置文档数上限，触发新索引的动态创建，显著简化管理流程。此外，配置中支持使用变量（如 {{ctx.source_index}}）动态生成目标索引名，便于多个任务复用同一模板，批量扩展 Rollup 任务时更加高效和灵活。

相比之下，OpenSearch Rollup 依赖 Index State Management（ISM）策略或手动操作进行索引切换，配置复杂、监控成本高，且在大规模任务部署时缺乏灵活的模板化机制。

• 差异：Easysearch 提供内建的自动滚动机制，OpenSearch 依赖外部策略或手动配置。
• 影响：Easysearch 更易于统一管理和大规模扩展，运维成本更低。

3. 聚合能力：更广泛的聚合类型支持

Easysearch Rollup 支持丰富的聚合类型，包括数值字段的 avg、sum、max、min、percentiles，关键词字段的 terms，日期字段的 date_histogram 与 date_range，还支持 filter 聚合与 special_metrics（可自定义聚合字段和方式）等高级功能。OpenSearch Rollup 支持的聚合类型相对有限，不支持 date_range、filter 等复杂聚合表达式。

• 差异：Easysearch 提供更全面的聚合能力，OpenSearch 仅支持基础聚合。
• 影响：Easysearch 更适合构建复杂的时序分析任务，满足更广泛的业务需求。

4. 性能优化：精细化配置与资源控制

Easysearch Rollup 提供多种性能优化选项，例如 ROLLUP_SEARCH_MAX_COUNT 控制并发查询数，rollup.hours_before 限制回溯时间范围，write_optimization 和 field_abbr 用于优化写入效率与运行时的内存占用。而 OpenSearch Rollup 缺乏类似的专用配置项，主要依赖通用的集群参数，灵活性与精细度较低。

• 差异：Easysearch 提供针对 Rollup 场景的专属优化选项，OpenSearch 优化能力较通用。
• 影响：Easysearch 在资源使用效率、查询性能和成本控制方面更具优势。

5. 任务管理：批量控制与更高灵活性

Easysearch Rollup 支持使用通配符进行任务批量管理，且新建任务默认处于非激活状态，用户可按需启用。而 OpenSearch Rollup 中，任务默认立即启用，管理粒度较粗，仅支持单个任务的启停与修改，缺乏批量操作能力。

• 差异：Easysearch 支持批量任务管理与按需启用，OpenSearch 功能较为基础。
• 影响：Easysearch 在多任务环境下提供更高的管理效率和控制能力。

实战示例：节点统计 Rollup 配置

以下是一个 Easysearch Rollup 任务的配置示例：

{
  "rollup": {
    "source_index": ".infini_metrics",
    "target_index": "rollup_node_stats_{{ctx.source_index}}",
    "timestamp": "timestamp",
    "continuous": true,
    "page_size": 200,
    "cron": "*/5 1-23 * * *",
    "interval": "1m",
    "identity": ["metadata.labels.cluster_id", "metadata.labels.node_id"],
    "stats": [{ "max": {} }, { "min": {} }, { "value_count": {} }],
    "special_metrics": [
      {
        "source_field": "payload.elasticsearch.node_stats.jvm.mem.heap_used_in_bytes",
        "metrics": [{ "avg": {} }, { "max": {} }, { "min": {} }]
      }
    ],
    "write_optimization": true
  }
}

• 亮点：支持自动索引滚动、标准 API 查询、special_metrics 高级聚合与写入优化等特性。

总结

综合来看，Easysearch Rollup 在以下方面优于 OpenSearch Rollup：

• 查询接口的兼容性与无感知集成
• 自动化的索引管理与扩展能力
• 更丰富的聚合类型与表达能力
• 针对性更强的性能优化参数
• 灵活高效的任务批量管理机制

这些优势使 Easysearch Rollup 更加适用于复杂、多样化的时序数据处理场景，特别是在对性能、扩展性与运维效率有较高要求的系统中表现出色。如果你正在寻找一款功能全面、易于管理的 Rollup 解决方案，Easysearch 是一个值得重点考虑的选择。

关于 Easysearch

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

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

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

之前介绍了 Easysearch 如何使用 S3 进行快照备份，毕竟那是手工操作。Easysearch 还提供了快照生命周期管理，能够按照策略自动创建、删除快照，极大地方便了用户的日常管理。

快照生命周期管理计划由创建计划、删除计划以及快照配置组成。

• 创建计划和删除计划包含一个 cron 表达式，指定任务的频率和时间。
• 删除计划可以指定快照保留策略，以保留过去 30 天的快照或仅保留最近的 10 个快照。
• 快照配置包括快照的索引和存储库，并支持所有通过 API 创建快照时的参数。此外，还可以指定快照名称中使用的日期的格式和时区。

快照生命周期创建的快照名称格式为 <policy _ name>-<date>-<Random number> 。

比如， 计划每 2 分钟对索引 .infini_metrics-00001 创建一个快照，并且只保留最近的 2 个快照。

curl -XPOST -uadmin:admin -H 'Content-Type: application/json' 'https://localhost:9200/_slm/policies/daily-policy' -d '
{
  "description": "测试快照策略",
  "creation": {
    "schedule": {
      "cron": {
        "expression": "*/2 * * * *",
        "timezone": "Asia/Shanghai"
      }
    },
    "time_limit": "1h"
  },
  "deletion": {
    "schedule": {
      "cron": {
        "expression": "*/1 * * * *",
        "timezone": "Asia/Shanghai"
      }
    },
    "condition": {
      "max_count": 2
    },
    "time_limit": "1h"
  },
  "snapshot_config": {
    "date_format": "yyyy-MM-dd-HH:mm",
    "date_format_timezone": "Asia/Shanghai",
    "indices": ".infini_metrics-00001",
    "repository": "easysearch_s3_repo",
    "ignore_unavailable": "true",
    "include_global_state": "false"
  }
}'

自动创建的快照如下图，一个 16 点 34 分创建的，另一个 16 点 36 分创建的。

⚠️ 注意：虽然指定只保留最近两个快照，但因为创建和删除其实是两个独立的任务，所以会短暂出现存在 3 个快照的现象，等删除任务调度一次就会删除多余的快照了。

如果遇到维护需要停止自动备份，也有相应的 API 来启停快照策略。

停止策略

curl -XPOST -uadmin:admin 'https://localhost:9200/_slm/policies/daily-policy/_start'

启动策略

curl -XPOST -uadmin:admin 'https://localhost:9200/_slm/policies/daily-policy/_stop'

查看策略

curl -XGET -uadmin:admin 'https://localhost:9200/_slm/policies'

删除策略

curl -XDELETE -uadmin:admin 'https://localhost:9200/_slm/policies/daily-policy?pretty'

更多详细信息请参考官方文档。

关于 Easysearch

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

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

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

Easysearch 内置了 S3 插件，这意味着用户可以直接使用该功能而无需额外安装任何插件。通过这一内置支持，用户能够方便快捷地执行 Amazon S3 上的数据快照操作。这种设计不仅简化了配置流程，也提高了工作效率，使得数据备份或迁移等任务变得更加简单易行。对于需要频繁与 S3 存储服务交互的应用场景来说，这是一个非常实用且高效的功能特性。

Minio

MinIO 是一款高性能的开源对象存储系统，专为存储大量的非结构化数据而设计。它提供了与 Amazon S3 兼容的 API，本次测试我们使用 MinIO 作为存储仓库。

建立 Bucket

进入 MinIO 管理界面，创建测试用的 bucket。

创建 Access key

测试的 Access Key 设置的比较简单。

Easysearch

为了能够使用 S3 存储，Easysearch 要进行必要的配置。

easyearch.yml

修改 easysearch.yml 配置 S3 信息。

s3.client.default.endpoint: 172.17.0.4:9000
s3.client.default.protocol: http

⚠️ 注意：修改了 easysearch.yml 需要重启生效。

keystore

为了安全，我们把 S3 的 Access key 信息加入 keystore 中。

bin/easysearch-keystore add s3.client.default.access_key #输入easysearch
bin/easysearch-keystore add s3.client.default.secret_key #输入easysearch
bin/easysearch-keystore list

注册存储库

在 INFINI Console 的开发工具中，使用命令注册 s3 存储库。

PUT /_snapshot/easysearch_s3_repo?verify=true&pretty
{
  "type": "s3",
  "settings": {
    "bucket": "easysearch-bucket",
    "compress": true
  }
}

更多参数请查看文档。

创建快照

在 INFINI Console 的开发工具中，使用命令创建快照。

备份执行完成。

S3 查看快照

我们在 INFINI Console 中通过命令创建了快照，可以在 MinIO 的 bucket 中进行进一步确认是否有相关文件。

快照还原测试

删除以备份索引 .infini_metrics-0001，删除前先查看下索引情况，文档数 557953。

删除 .infini_metrics-0001 索引。

确认 .infini_metrics-0001 索引已被删除。

进行快照还原。

验证恢复索引。

索引 .infini_metrics-0001 已经还原了，文档数也一致。

小结

Easysearch 使用 S3 存储备份的步骤如下：

1. S3 服务建立 Bucket、Access Key。
2. Easysearch 编辑 easysearch.yml 添加 S3 服务 endpoint 信息。
3. easysearch-keystore 添加 S3 的 Access key 信息，加密保存。
4. Easysearch 注册 S3 存储仓库。
5. 执行快照备份。

关于 Easysearch

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

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

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

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 在日志处理场景中具有更高的性能与存储效率优势，尤其适用于大规模分片与海量数据的使用场景。

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