DataX

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

本篇主要介绍 DataX 如何将数据写入到 Easysearch，对于各种数据源的连接不会做深入的探讨，感兴趣的小伙伴可以访问 [DataX](https://github.com/alibaba/Dat ... ion.md) 的 Github 仓库查看详情。

下载与安装

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

系统需求：

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

   创建任务配置文件
   

   
   每个数据同步的操作可称为一个任务，任务的配置文件定义了数据源(reader)、数据目的(writer) ，以及任务的设置信息，如并发数、速度控制等。DataX 集成了如此多的数据源，如果靠纯手工编写任务配置显然不现实。官方也出了个命令可以根据指定的数据源和数据目的帮助大家生成任务配置。
   
   shell<br /> python datax.py -r {YOUR_READER} -w {YOUR_WRITER}<br />
   
   测试配置文件
   
   此次演示使用 streamreader 和 elasticsearchwriter 作为数据源和数据目的，任务配置如下：
   
   json<br /> {<br /> "job": {<br /> "content": [<br /> {<br /> "reader": {<br /> "name": "streamreader",<br /> "parameter": {<br /> "sliceRecordCount": 10000,<br /> "column": [<br /> {<br /> "type": "long",<br /> "value": "10"<br /> },<br /> {<br /> "type": "string",<br /> "value": "hello，你好，世界-DataX"<br /> },<br /> {<br /> "type": "string",<br /> "value": "hello，你好，Easysearch"<br /> }<br /> ]<br /> }<br /> },<br /> "writer": {<br /> "name": "elasticsearchwriter",<br /> "parameter": {<br /> "endpoint": "<a href="http://localhost:9200"" rel="nofollow" target="_blank">http://localhost:9200"</a>,<br /> "accessId": "admin",<br /> "accessKey": "1ef0c661d8562aaa06be",<br /> "index": "yf-test",<br /> "column": [<br /> { "name": "no", "type": "long" },<br /> { "name": "content", "type": "keyword" },<br /> { "name": "content2", "type": "keyword" }<br /> ]<br /> }<br /> }<br /> }<br /> ],<br /> "setting": {<br /> "speed": {<br /> "channel": 50<br /> }<br /> }<br /> }<br /> }<br />
   
   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](https://infinilabs.cn/docs/latest/gateway/) 代理 ES 服务，提供 http 通道给离线数据同步专用。
     
     ⚠️ 注意：
     
     不同的 reader、writer 对 sliceRecordCount 和 channel 会有不同的行为。
     
     

     Easysearch
     

     
     本次测试使用的 [Easysearch](https://infinilabs.cn/download/) 版本是 1.9.0，需要注意是 Easysearch 要开启兼容性参数：
     
     yml<br /> elasticsearch.api_compatibility: true<br />
     
     否则创建索引报错退出。(实际索引创建成功了但是 mapping 信息是空的)
     
     
     
     

     运行任务
     

     
     编辑好任务配置文件后，下一步就是执行任务。
     
     shell<br /> python3 datax.py yf-test.json<br />
     
     
     
     写入数据时索引不存在，Datax 根据 schema 定义创建了索引。
     
     
     
     OK 任务执行完毕，写入 50w 个文档耗时 10 秒。
     
     如果有其他问题欢迎与我联系。
     
     

本文将探讨如何使用 INFINI Gateway 阻止不合理的查询发送到 Elasticsearch，此方法同样适用于 Opensearch 和 [Easysearch](https://infinilabs.cn/products/easysearch/) 。

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

具体来说：

• 查询吞吐量过高：当查询请求的频率或复杂度超过了集群的处理能力时，可能会导致节点内存耗尽，从而引发 OOM。
  
• 执行不合理查询：某些特殊类型的查询（例如涉及大量嵌套、深度分页或复杂的聚合操作）可能需要大量的内存资源，在执行过程中也可能导致 OOM。
  
  通过识别并优化这些查询模式，可以有效减少 OOM 事件的发生。针对查询吞吐量过高的情况，可以参考之前的[文章](https://infinilabs.cn/blog/202 ... s-oom/)来管理查询吞吐。接下来的内容将介绍如何阻挡不合理查询，保护集群稳定。
  
  

  不合理查询
  

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

• 嵌套聚合查询
  
• 使用复杂的正则表达式进行模糊匹配
  
• 深度分页查询（如 from: 10000, size: 10）
  
• 脚本查询（Script Query）
  
• 大规模嵌套聚合查询
  
  为了防止这些查询对 Elasticsearch 集群造成影响，我们可以使用 [INFINI Gateway](https://infinilabs.cn/products/gateway/) 对这些查询进行阻断。
  
  

  请求上下文
  

  
  INFINI Gateway 运行环境中有非常多的信息可被利用，而请求上下文就是访问这些信息的入口。如请求来源、请求体信息等，都可使用关键字 _ctx 作为前缀访问相应的上下文信息。
  
  HTTP 请求内置的 _ctx 上下文对象主要包括如下：
  
  
  
  更多的上下文信息请访问[文档](https://infinilabs.cn/docs/lat ... ntext/)。
  
  

  context_filter
  

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

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

  使用步骤
  

  
  

  1. 确定关键字信息：确定特殊查询请求中的关键特征或关键字。
     
  2. 配置匹配规则：在 context_filter 中定义相应的匹配规则，选择合适的匹配模式（如前缀、后缀、模糊或正则匹配）。
     
  3. 阻断请求：一旦匹配到这些关键字，INFINI Gateway 将自动阻断请求并返回指定的消息。
     
     更多详细内容，请参阅相关 [文档](https://docs.infinilabs.com/ga ... ilter/)。
     
     

     举个例子
     

     
     阻止 wildcard 查询（模糊匹配查询），我们先看一个 wildcard 查询的样子。
     
     plain<br /> GET yf-test-1shard/_search<br /> {<br /> "query": {<br /> "wildcard": {<br /> "path.keyword": {<br /> "value": "/a*"<br /> }<br /> }<br /> }<br /> }<br />
     
     上面的查询，会查询 path 字段，所有以 /a 开头的文档。
     
     
     
     第一步：我们可确定关键字是 wildcard，为了进一步限制是 wildcard 查询里的情况，我们可将关键字确定为 wildcard":，因为有时候查询 url 里会有 expand_wildcards 字样。
     
     第二步：编辑 INFINI Gateway 默认配置文件，增加 context_filter 匹配规则。
     
     ```yaml
     

• 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":'
      ``<br /> <br /> 通过上面的修改，我们在 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](https://infinilabs.cn/products/easysearch/) 从 1.8.0 版本开始引入了写入限速功能，靠引擎自身能力对写入速度进行限制。不仅听着简单，实际用起来一点也不麻烦，我们赶紧实战一把。

测试环境

INFINI Easyssearch 1.9.0，单节点集群。

测试方法

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

plain<br /> ./loadgen-linux-amd64 -config ../config/write-yf-test.yml -d 3000 -l 5000<br />

请求示例

plain<br /> {"index": {"_index": "yf-test-1shard","_id": "$[[uuid]]"}}<br /> {"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"}<br />

测试基线

单节点不限速写入测试

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


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


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


节点级别限速

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

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

压测单个索引，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 进行限速。在索引的设置里配置相应的限流阈值：

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

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


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


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

分片级别限速

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

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

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

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


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

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


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

注意事项

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

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

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

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

## 背景

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


## 使用在线工具 certificatetools.com

在允许生成自签名证书的在线服务中，[CertificateTools](https://certificatetools.com/) 是最先进的。只需查看所有可用选项即可：


就这么简单！

## 使用 Let’s Encrypt

首先，安装 Certbot，这是 [Let’s Encrypt](https://letsencrypt.org/) 官方推荐的工具，用于自动化获取和续期 SSL/TLS 证书。

**1. 安装 Certbot**

- 访问 [Certbot 下载页面](https://certbot.eff.org/instructions)。
- 选择 **Windows** 系统，下载并安装 Certbot。

**2. 获取证书**

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

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

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

```powershell
C:\Certbot\live\example.com\
```

里面有以下文件：

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

**3. 导入证书**

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

**4. 验证证书**

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

**5. 续期证书**

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

```powershell
certbot renew
```

## 使用 OpenSSL

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

**步骤：**

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

```powershell
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 窗口中输入以下命令：

```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 官方网站](https://www.hohnstaedt.de/xca/) 下载并安装 XCA。
2. 启动 XCA，点击 文件 > 新建数据库 来创建一个新的证书数据库。
3. 在 证书 选项卡中，点击 新建证书。
4. 在证书的设置中，选择 自签名证书，然后填写证书的详细信息。
5. 点击 保存 来生成证书。

**验证：**

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

## 总结

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

---

**参考资料**

1. [7 ways to create self-signed certificates on Windows
](https://passwork.pro/blog/7-wa ... ndows/)

2. [如何使用 XCA 创建证书](https://campus.barracuda.com/p ... h-xca/)

[Easysearch](https://infinilabs.cn/products/easysearch/) 的很多用户都有这样的需要，之前是用的 ES，现在要迁移到 Easysearch，但是业务方使用的是 Spring Boot 集成的客户端，问是否能平滑迁移。

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

服务器设置

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

```yml
elasticsearch.api_compatibility: true

根据客户端版本配置版本号，我这里配置成 7.17.18

elasticsearch.api_compatibility_version: "7.17.18"
```

项目设置

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

```xml

11
4.4.18
7.17.18




org.springframework.boot
spring-boot-starter-web



org.springframework.data
spring-data-elasticsearch
${spring-data-elasticsearch.version}


org.elasticsearch
elasticsearch
${elasticsearch.version}


org.elasticsearch.client
elasticsearch-rest-high-level-client
${elasticsearch.version}


org.projectlombok
lombok
true


```

## 客户端连接配置

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

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

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

连接配置类

```java
@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 注解定义领域模型：

```java
@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 的仓库接口：

```java
@Repository
@EnableElasticsearchRepositories
public interface ProductRepository extends ElasticsearchRepository {
}
```

## 服务层

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

```java
@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);
}
}
```

## 测试

编写集成测试类：

```java
@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 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 queries = products.stream()
.map(product -> new IndexQueryBuilder()
.withObject(product)
.withIndex(INDEX_NAME)
.build())
.collect(Collectors.toList());

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

// 验证结果
List 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/202 ... tion/

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

通过查看 Easysearch [用户](https://infinilabs.cn/docs/lat ... pi%23创建用户)接口文档，创建用户使用如下接口：

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

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

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

API 接口创建的用户

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


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


配置文件创建的用户

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

```yaml

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"
  ``<br /> <br /> 其中hash字段是将用户密码哈希出来的值，可通过bin/hash_password.sh -p "<明文密码>"生成。<br /> <br /> 进行用户访问验证。<br /> <br /> ![](<a href="https://infinilabs.cn/img/blog/2025/easysearch-use-api-reset-user-password/3.pn" rel="nofollow" target="_blank">https://infinilabs.cn/img/blog ... /3.pn</a>g)<br /> <br /> 对 test1、test2 用户进行密码重置。<br /> <br /> ![](<a href="https://infinilabs.cn/img/blog/2025/easysearch-use-api-reset-user-password/4.pn" rel="nofollow" target="_blank">https://infinilabs.cn/img/blog ... /4.pn</a>g)<br /> <br /> 用户 test1 重置失败，用户 test2 重置成功。用户 test1 重置失败是因为配置了reserved: true，将用户 test1 指定为内置用户，使用用户 admin 会因为权限无法进行密码重置，那就需要使用有更高权限的[管理证书](<a href="https://infinilabs.cn/docs/latest/easysearch/references/security/configuration/tls#" rel="nofollow" target="_blank">https://infinilabs.cn/docs/lat ... ls%23</a>配置管理证书)进行密码重置。默认在 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 的兼容性参数有两个，我们先开第一个。

```plain
elasticsearch.api_compatibility: true

elasticsearch.api_compatibility_version: "8.9.0"

<br /> <br /> 启动 Filebeat 连接。<br /> <br /> ![](<a href="https://infinilabs.cn/img/blog/2025/filebeat-8-connect-easysearch/2.pn" rel="nofollow" target="_blank">https://infinilabs.cn/img/blog ... /2.pn</a>g)<br /> <br /> 这次 Filebeat 检测到了一个兼容的版本 7.10.2，并导入了一个叫 filebeat 的索引生命周期管理策略，但最终因为无法导入模板而失败了。<br /> <br /> 索引生命周期查看。<br /> <br /> ![](<a href="https://infinilabs.cn/img/blog/2025/filebeat-8-connect-easysearch/3.pn" rel="nofollow" target="_blank">https://infinilabs.cn/img/blog ... /3.pn</a>g)<br /> <br /> **关闭 Filebeat 的导入模板功能尝试连接**，修改配置 filebeat.yml 关闭模板导入。<br /> <br /> ![](<a href="https://infinilabs.cn/img/blog/2025/filebeat-8-connect-easysearch/4.pn" rel="nofollow" target="_blank">https://infinilabs.cn/img/blog ... /4.pn</a>g)<br /> <br /> plain
setup.template.enabled: false
```

启动 Filebeat 连接 Easysearch。


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


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

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

plain<br /> elasticsearch.api_compatibility: true<br /> elasticsearch.api_compatibility_version: "8.9.0"<br />

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


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

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


数据也成功写入。


注意事项

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

如果 Easysearch 使用的是 1.9.0 之前的版本，可以使用[网关代理 Easysearch 的方法](https://infinilabs.cn/blog/202 ... earch/)。

总结

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 部分
   
   ```plain
   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
   
   ```plain
   
   • 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 的兼容参数（默认已注释）：
         
         plain<br /> elasticsearch.api_compatibility: true<br /> elasticsearch.api_compatibility_version: "8.9.0"<br />
         
         注意 Logstash 配置中关闭自动导入模板功能。
         
         
         
         plain<br /> output {<br /> elasticsearch {<br /> hosts => ["<a href="http://127.0.0.1:9200"" rel="nofollow" target="_blank">http://127.0.0.1:9200"</a>]<br /> index => "logstash8"<br /> manage_template => false<br /> }<br /> }<br />
         
         启动 Logstash 连接到 easysearch 后， Logstash 日志会输出检测到版本 8.9.0 。
         
         
         
         最新 Easysearch 下载地址：[下载](https://infinilabs.cn/download/)
         
         好了，还有其他问题的小伙伴，欢迎加我微信沟通。
         

压测环境

虚拟机配置

使用阿里云上规格：[ecs.u1-c1m4.4xlarge](https://help.aliyun.com/docume ... l%23u1)，PL2: 单盘 IOPS 性能上限 10 万 (适用的云盘容量范围：461GiB - 64TiB)

| vCPU | 内存 (GiB) | 磁盘(GB) | 带宽（Gbit/s） | 数量 |
| ---- | ---------- | -------- | -------------- | ---- |
| 16 | 64 | 500 | 5000 | 24 |

Easysearch 配置

7 节点集群，版本：1.9.0

| 实例名 | 内网 IP | 软件 | vCPU | JVM | 磁盘 |
| ---------------------- | -------------- | ---------- | ---- | ---- | ----- |
| i-2zegn56cijnzklcn2410 | 172.22.75.144 | Easysearch | 16 | 31G | 500GB |
| i-2zegn56cijnzklcn240u | 172.23.15.97 | Easysearch | 16 | 31G | 500GB |
| i-2zegn56cijnzklcn240i | 172.25.230.228 | Easysearch | 16 | 31G | 500GB |
| i-2zegn56cijnzklcn240y | 172.22.75.142 | Easysearch | 16 | 31G | 500GB |
| i-2zegn56cijnzklcn240x | 172.22.75.143 | Easysearch | 16 | 31G | 500GB |
| i-2zegn56cijnzklcn240z | 172.24.250.252 | Easysearch | 16 | 31G | 500GB |
| i-2zegn56cijnzklcn240r | 172.24.250.254 | Easysearch | 16 | 31G | 500GB |

Elasticsearch 配置

7 节点集群，版本：7.10.2

| 实例名称 | 内网 IP | 软件 | vCPU | JVM | 磁盘 |
| ---------------------- | -------------- | ------------- | ---- | ---- | ----- |
| i-2zegn56cijnzklcn240m | 172.24.250.251 | Elasticsearch | 16 | 31G | 500GB |
| i-2zegn56cijnzklcn240p | 172.22.75.145 | Elasticsearch | 16 | 31G | 500GB |
| i-2zegn56cijnzklcn240o | 172.17.67.246 | Elasticsearch | 16 | 31G | 500GB |
| i-2zegn56cijnzklcn240t | 172.22.75.139 | Elasticsearch | 16 | 31G | 500GB |
| i-2zegn56cijnzklcn240q | 172.22.75.140 | Elasticsearch | 16 | 31G | 500GB |
| i-2zegn56cijnzklcn240v | 172.24.250.253 | Elasticsearch | 16 | 31G | 500GB |
| i-2zegn56cijnzklcn240l | 172.24.250.250 | Elasticsearch | 16 | 31G | 500GB |

监控集群配置

单节点 Easysearch 集群，版本：1.9.0

| 实例名 | 内网 IP | 软件 | vCPU | 内存 | 磁盘 |
| ---------------------- | -------------- | -------------------- | ---- | ---- | ----- |
| i-2zegn56cijnzklcn240f | 172.25.230.226 | 监控集群：Console | 16 | 64G | 500GB |
| i-2zegn56cijnzklcn240j | 172.23.15.98 | 监控集群：Easysearch | 16 | 64G | 500GB |

压测 loadgen 配置

loadgen 版本：1.25.0

4 台压 Easysearch，4 台压 Elasticsearch。

| 实例名 | 内网 IP | 软件 | vCPU | 内存 | 磁盘 |
| ---------------------- | -------------- | -------------------------- | ---- | ---- | ----- |
| i-2zegn56cijnzklcn240n | 172.17.67.245 | Loadgen - 压 Easysearch | 16 | 64G | 500GB |
| i-2zegn56cijnzklcn2411 | 172.22.75.141 | Loadgen - 压 Easysearch | 16 | 64G | 500GB |
| i-2zegn56cijnzklcn240k | 172.25.230.227 | Loadgen - 压 Easysearch | 16 | 64G | 500GB |
| i-2zegn56cijnzklcn240e | 172.22.75.138 | Loadgen - 压 Easysearch | 16 | 64G | 500GB |
| i-2zegn56cijnzklcn240h | 172.24.250.255 | Loadgen - 压 Elasticsearch | 16 | 64G | 500GB |
| i-2zegn56cijnzklcn240w | 172.24.251.0 | Loadgen - 压 Elasticsearch | 16 | 64G | 500GB |
| i-2zegn56cijnzklcn240g | 172.24.250.248 | Loadgen - 压 Elasticsearch | 16 | 64G | 500GB |
| i-2zegn56cijnzklcn240s | 172.24.250.249 | Loadgen - 压 Elasticsearch | 16 | 64G | 500GB |

压测索引 Mapping

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

压测方法

每 4 个 [loadgen](https://infinilabs.cn/docs/lat ... hmark/) 使用批量写入接口 bulk 轮询压测同一集群的 7 个节点，每个请求写入 10000 个文档。

具体请求如下：

```plain
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]]
  ```
  
  

  压测数据样本
  

  
  plain<br /> {"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}<br /> {"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}<br /> {"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; +<a href="http://www.google.com/bot.htm" rel="nofollow" target="_blank">http://www.google.com/bot.htm</a>l)","connections":500,"timestamp":"2024-11-16T15:14:30.424","status":403}<br />
  
  

  压测索引 1 主分片 0 副本
  

  
  

  Elastic 吞吐
  

  
  
  
  
  
  

  Elastic 线程及队列
  

  
  
  
  

  资源消耗
  

  
  
  
  
  
  
  
  

  Easysearch 吞吐
  

  
  
  
  
  
  

  Easysearch 线程及队列
  

  
  
  
  

  资源消耗
  

  
  
  
  
  
  
  
  

  对比
  

  
  | 软件 | 平均集群吞吐 | 平均单节点吞吐 | 最大队列 | 磁盘消耗 |
  | ------------- | ------------ | -------------- | -------- | -------- |
  | Elasticsearch | 5w | 5w | 811 | 10G |
  | Easysearch | 7w | 7w | 427 | 4G |
  
  

  压测索引 1 主分片 1 副本
  

  
  

  Elastic 吞吐
  

  
  
  
  
  
  

  Elastic 线程及队列
  

  
  
  
  

  资源消耗
  

  
  
  
  

  Easysearch 吞吐
  

  
  
  
  
  
  

  Easysearch 线程及队列
  

  
  
  
  

  资源消耗
  

  
  
  
  

  对比
  

  
  | 软件 | 平均集群吞吐 | 平均单节点吞吐 | 最大队列 | 磁盘消耗(~3000 万文档） |
  | ------------- | ------------ | -------------- | -------- | ----------------------- |
  | Elasticsearch | 10w | 5w | 791 | 22G |
  | Easysearch | 14w | 7w | 421 | 7G |
  
  

  压测索引 7 主分片
  

  
  

  Elastic 吞吐
  

  
  
  
  
  
  

  Elastic 线程及队列
  

  
  
  
  

  资源消耗
  

  
  
  
  
  
  网络
  
  单节点平均接收 26MB/s，对应带宽：1456 Mb/s
  
  
  
  5 千万文档，总存储 105 GB，单节点 15 GB
  
  
  
  

  Easysearch 吞吐
  

  
  
  
  
  
  

  Easysearch 线程及队列
  

  
  
  
  

  资源消耗
  

  
  
  
  
  
  
  
  
  
  

  对比
  

  
  | 软件 | 平均集群吞吐 | 平均单节点吞吐 | 最大队列 | 磁盘消耗 |
  | ------------- | ------------ | -------------- | -------- | -------- |
  | Elasticsearch | 35w | 5w | 2449 | 105G |
  | Easysearch | 60w | 8.5w | 1172 | 36G |
  
  

  总结
  

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

  • Easysearch 相比 Elasticsearch 的索引性能显著提升
    Easysearch 集群的吞吐性能提升了 40% - 70%，且随着分片数量的增加，性能提升效果更为显著。
    
  • Easysearch 相比 Elasticsearch 的磁盘压缩效率大幅提高
    Easysearch 集群的磁盘压缩效率提升了 2.5 - 3 倍，并且随着数据量的增加，压缩效果愈发明显。
    
    此测试结果表明，[Easysearch](https://infinilabs.cn/docs/latest/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），并将结果存储到目标索引中。
     
     

     示例：
     

     
     json<br /> "metrics": [<br /> "payload.elasticsearch.index_stats.*"<br /> ]<br />
     
     在这个例子中，metrics 指定了对 payload.elasticsearch.index_stats 下的所有字段进行聚合。
     
     ---
     
     

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

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

     示例：
     

     
     json<br /> "attributes": [<br /> "agent.*",<br /> "metadata.*"<br /> ]<br />
     
     在这个例子中，attributes 指定了所有以 agent. 和 metadata. 开头的字段都会被保留到 Rollup 结果中。
     
     ---
     
     

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

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

     示例：
     

     
     json<br /> "exclude": ["payload.elasticsearch.index_stats.routing.*"]<br />
     
     在这个例子中，exclude 指定了 payload.elasticsearch.index_stats.routing 下的所有字段都不会被处理。
     
     ---
     
     

     4. filter 参数：过滤源数据
     

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

     示例：
     

     
     json<br /> "filter": {<br /> "metadata.name": "index_stats"<br /> }<br />
     
     在这个例子中，filter 指定了只有 metadata.name 字段值为 index_stats 的文档才会被 Rollup 任务处理。
     
     ---
     
     

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

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

     功能：
     

     
     

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

     示例：
     

     
     json<br /> "identity": [<br /> "metadata.labels.cluster_id",<br /> "metadata.labels.index_id",<br /> "metadata.labels.index_name",<br /> "metadata.category",<br /> "payload.elasticsearch.index_stats.index_info.health"<br /> ]<br />
     
     在这个例子中，identity 指定了多个字段作为分组条件。Rollup 任务会根据这些字段的值对数据进行分组，然后在每个分组内进行聚合计算。
     
     ---
     
     

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

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

     功能：
     

     
     

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

     示例：
     

     
     json<br /> "interval": "1m"<br />
     
     在这个例子中，interval 指定了时间聚合的间隔为 1 分钟。Rollup 任务会每分钟对数据进行一次聚合。
     
     ---
     
     

     1.10 版本的新特性
     

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

     1. 支持 date_range 聚合
     

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

     示例：
     

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

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

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

     示例：
     

     
     json<br /> POST _rollup/jobs/rollup*/_start<br /> POST _rollup/jobs/rollup*/_stop<br />
     
     ---
     
     

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

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

     示例：
     

     
     json<br /> PUT /_cluster/settings<br /> {<br /> "transient": {<br /> "rollup.max_docs": 10000000<br /> }<br /> }<br />
     
     ---
     
     

     4. 新增 ROLLUP_SEARCH_MAX_COUNT 配置
     

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

     功能：
     

     
     

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

     示例：
     

     
     json<br /> PUT /_cluster/settings<br /> {<br /> "transient": {<br /> "rollup.search.max_count": 2<br /> }<br /> }<br />
     
     在这个例子中，ROLLUP_SEARCH_MAX_COUNT 被设置为 2，表示 Rollup 任务在执行搜索请求时，最多会同时发送 2 个并发分片请求。
     
     

     配置建议：
     

     
     

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

     创建 Rollup 任务的完整示例
     

     
     以下是一个完整的 Rollup 任务配置示例，展示了 metrics、attributes、exclude 和 filter 参数的综合使用：
     
     auto<br /> PUT _rollup/jobs/rollup1<br /> {<br /> "rollup": {<br /> "source_index": ".infini_metrics",<br /> "target_index": "rollup1_{{ctx.source_index}}",<br /> "timestamp": "timestamp",<br /> "continuous": true,<br /> "page_size": 1000,<br /> "cron": "*/10 1-23 * * *",<br /> "timezone": "UTC+8",<br /> "stats": [<br /> {<br /> "max": {}<br /> },<br /> {<br /> "value_count": {}<br /> }<br /> ],<br /> "interval": "1m",<br /> "identity": [<br /> "metadata.labels.cluster_id",<br /> "metadata.labels.index_id",<br /> "metadata.labels.index_name",<br /> "metadata.category",<br /> "payload.elasticsearch.index_stats.index_info.health"<br /> ],<br /> "attributes": [<br /> "agent.*",<br /> "metadata.*"<br /> ],<br /> "metrics": [<br /> "payload.elasticsearch.index_stats.*"<br /> ],<br /> "exclude": ["payload.elasticsearch.index_stats.routing.*"],<br /> "filter": {<br /> "metadata.name": "index_stats"<br /> }<br /> }<br /> }<br />
     
     ---
     
     

     如何使用 Rollup 索引
     

     
     从 1.10.0 版本开始，索引生命周期插件不再默认启用 rollup 搜索功能，如果想使用搜索 rollup 搜索功能，需要设置
     
     auto<br /> PUT /_cluster/settings<br /> {<br /> "transient": {<br /> "rollup.search.enabled": true<br /> }<br /> }<br /> <br />
     
     无需特意搜索 rollup 索引，只需使用标准的 _search API 对原始目标索引进行搜索。需要注意的是，查询时必须符合目标索引的约束条件。
     
     以下是一个使用 Rollup 索引的示例：
     
     auto<br /> GET target-test/_search<br /> {<br /> "size": 0,<br /> "aggs": {<br /> "a": {<br /> "date_histogram": {<br /> "field": "@timestamp",<br /> "fixed_interval": "1h"<br /> }<br /> },<br /> "total_passenger_count": {<br /> "sum": {<br /> "field": "passenger_count"<br /> }<br /> }<br /> }<br /> }<br />
     
     ---
     
     

     总结
     

     
     Rollup 是处理时序数据的强大工具，能够有效降低存储成本并提高查询性能。Easysearch 的 Rollup 功能不仅支持多种聚合类型，还提供了自动滚动索引、无缝查询等特性，极大地简化了时序数据的管理和分析流程。通过合理配置 metrics、attributes、exclude 和 filter 参数，你可以灵活地控制 Rollup 任务的行为，从而高效地处理时序数据。
     
     如果你正在处理大量的时序数据，不妨尝试使用 Rollup 来优化你的数据存储和查询。通过本文的介绍，相信你已经对 Rollup 有了深入的了解。赶快动手试试吧，体验 Rollup 带来的高效与便捷！
     
     更详细的使用文档可在 [官网](https://www.infinilabs.cn/docs ... p_api/) 查看
     
     

     关于 Easysearch
     

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

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

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

迁移路径

| Elasticsearch 版本 | 快照兼容 | 推荐升级方法 |
| ---------------------- | ------------ | ------------------------ |
| 5.x | ❌ | 使用 INFINI Console 迁移 |
| 6.x | ✅ | 快照恢复迁移 |
| 7.0.0 - 7.10.2 | ✅ | 快照恢复迁移 |
| >7.11.0 | ❌ | 使用 INFINI Console 迁移 |

之前有同事做过相关测试，详情请移步[这里](https://infinilabs.cn/blog/202 ... store/)。

快照恢复迁移

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

    INFINI Console 迁移
    

    
    

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

    客户端调整
    

    
    如果要继续使用 Java High Level REST Client，建议将版本调整到 7.10.2 。当然更建议的是使用 Easysearch 的[客户端](https://infinilabs.cn/blog/202 ... ent-1/)，更轻更快，构建查询，跟搭积木一样简单。
    
    

    开源事业
    

    
    [极限科技（INFINI Labs）](https://infinilabs.cn) 一直致力于为开发者和企业提供优质的开源工具，提升整个技术生态的活力。除了维护国内最流行的分词器 analysis-ik 和 analysis-pinyin ，也在不断推动更多高质量开源产品的诞生。
    
    在极限科技成立三周年之际，公司宣布以下产品和工具已全面开源：
    
    

    • [INFINI Framework](https://github.com/infinilabs/framework)
      
    • [INFINI Gateway](https://github.com/infinilabs/gateway)
      
    • [INFINI Console](https://github.com/infinilabs/console)
      
    • [INFINI Agent](https://github.com/infinilabs/agent)
      
    • [INFINI Loadgen](https://github.com/infinilabs/loadgen)
      
    • [INFINI Coco AI](https://github.com/infinilabs/coco-app)
      
      以上开源软件都可以在 Github 上面找到：[https://github.com/infinilabs](https://github.com/infinilabs)
      
      欢迎大家一起参与到开源工具的维护、贡献当中来，别忘了 Star🌟 支持一下！！！
      
      如果您对迁移过程有任何疑问，欢迎与我讨论。
      
      

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

角色设置

相信你对节点角色的概念已经有所熟悉。要启用可搜索快照功能，[Easysearch](https://infinilabs.cn/products/easysearch/) 集群中必须至少有一个节点拥有 search 角色。参考设置如下。

plain<br /> node.roles: ["search"]<br /> node.search.cache.size: 500mb<br />

• node.roles: 指定节点角色，只有 search 角色的节点才能去搜索快照中的数据。
  
• node.search.cache.size: 执行快照搜索时，数据缓存大小。
  
  混合角色设置，参考如下。
  
  plain<br /> node.roles: ["master","data","search","ingest"]<br /> node.search.cache.size: 500mb<br />
  
  

  创建快照
  

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

  创建 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 存储类型来创建快照索引。
  
  创建快照索引时，注意名称不能与当前索引名称重复。通常我们备份完索引后，可删除索引释放节点磁盘空间，创建快照索引时默认使用原来的名称。
  
  ```plain
  

  删除 infini 索引释放磁盘空间
  

  DELETE infini
  
  

  创建快照索引,使用原索引名称
  

  POST /_snapshot/my-fs-repository/1/_restore
  {
  "indices": "infini",
  "include_global_state": false,
  "include_aliases": false,
  "storage_type": "remote_snapshot"
  }
  <br /> <br /> 创建快照索引的命令和还原快照的命令非常相似，关键在于 storage_type 参数指定 remote_snapshot 存储类型。<br /> <br /> 如果要将快照中的全部索引都创建快照索引，可省略 indices 参数。<br /> <br /> ![](<a href="https://infinilabs.cn/img/blog/2024/searchable-snapshot-read-this-article/2.pn" rel="nofollow" target="_blank">https://infinilabs.cn/img/blog ... /2.pn</a>g)<br /> <br /> 如果想在创建快照索引时指定不同的名字，参考下面的命令。<br /> <br /> plain
  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 信息，普通索引没有此信息。
  
  
  
  快照索引能写入数据吗？
  
  快照索引无法写入，数据仍然保持在快照格式中存储在存储库中，因此可搜索快照索引本质上是只读的。 任何尝试写入可搜索快照索引的操作都会导致错误。
  
  
  
  快照索引不想要了怎么办？
  
  直接删除，需要时再执行创建快照索引流程。此外快照在创建快照索引后，无法直接删除快照，要先删除快照索引。
  
  
  如果您对上述内容有任何疑问，欢迎与我讨论。
  
  

[Easysearch](https://infinilabs.cn/products/easysearch/) 的很多用户都有这样的需要，之前是用的 ES，现在要迁移到 Easysearch，但是业务方使用的是 Spring Boot 集成的客户端，问是否能平滑迁移。

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

服务器设置

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

```yml
elasticsearch.api_compatibility: true

根据客户端版本配置版本号，我这里配置成 7.17.18

elasticsearch.api_compatibility_version: "7.17.18"
```

项目设置

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

```xml

11
4.4.18
7.17.18




org.springframework.boot
spring-boot-starter-web



org.springframework.data
spring-data-elasticsearch
${spring-data-elasticsearch.version}


org.elasticsearch
elasticsearch
${elasticsearch.version}


org.elasticsearch.client
elasticsearch-rest-high-level-client
${elasticsearch.version}


org.projectlombok
lombok
true


```

## 客户端连接配置

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

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

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

连接配置类

```java
@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 注解定义领域模型：

```java
@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 的仓库接口：

```java
@Repository
@EnableElasticsearchRepositories
public interface ProductRepository extends ElasticsearchRepository {
}
```

## 服务层

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

```java
@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);
}
}
```

## 测试

编写集成测试类：

```java
@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 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 queries = products.stream()
.map(product -> new IndexQueryBuilder()
.withObject(product)
.withIndex(INDEX_NAME)
.build())
.collect(Collectors.toList());

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

// 验证结果
List 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/202 ... tion/


各位 [Easysearch](https://infinilabs.cn/docs/latest/easysearch/) 的小伙伴们，我们前一阵刚把 easysearch-client 更新到了 2.0.2 版本！借此详细介绍下新版客户端的使用。

新版客户端和 1.0 版本相比，完全重构，抛弃了旧版客户端的一些历史包袱，从里到外都焕然一新！不管是刚入门的小白还是经验丰富的老司机，2.0.x 客户端都能让你开发效率蹭蹭往上涨！

到底有啥新东西？

• 更轻更快： 以前的版本依赖了一堆乱七八糟的东西，现在好了，我们把那些没用的都砍掉了，客户端变得更苗条，性能也杠杠的！
  
• 类型安全，告别迷糊： 常用的 Easysearch API 现在都配上了强类型的请求和响应对象，再也不用担心写错参数类型了，代码也更好看了，维护起来也更省心！
  
• 同步异步，想咋用咋用： 所有 API 都支持同步和异步两种调用方式，不管是啥场景，都能轻松应对！
  
• 构建查询，跟搭积木一样简单： 我们用了流式构建器和函数式编程，构建复杂查询的时候，代码写起来那叫一个流畅，看着也舒服！
  
• 和 Jackson 无缝对接： 可以轻松地把你的 Java 类和客户端 API 关联起来，数据转换嗖嗖的快！
  
  

  快速上手
  

  
  废话不多说，咱们直接上干货！这部分教你怎么快速安装和使用 easysearch-client 2.0.2 客户端，还会演示一些基本操作。
  
  

  安装
  

  
  easysearch-client 2.0.2 已经上传到 Maven 中央仓库了，加到你的项目里超级方便。
  
  最低要求： JDK 8 或者更高版本
  
  依赖管理： 客户端内部用 Jackson 来处理对象映射。
  
  

  Maven 项目
  

  
  在你的 pom.xml 文件的 <dependencies> 里面加上这段：
  
  ```xml
  

  
  
  com.infinilabs
  easysearch-client
  2.0.2
  
  
  ```
  
  #### Gradle 项目
  
  在你的 `build.gradle` 文件的 `dependencies` 里面加上这段：
  
  ```gradle
  dependencies {
  implementation 'com.infinilabs:easysearch-client:2.0.2'
  }
  ```
  
  ### 初始化客户端
  
  下面这段代码演示了怎么初始化一个启用了安全通信加密和 security 的 Easysearch 客户端，看起来有点长，别慌，我们一步一步解释！
  
  ```java
  public static EasysearchClient create() throws NoSuchAlgorithmException, KeyStoreException,
  KeyManagementException {
  
  final HttpHost[] hosts = new HttpHost[]{new HttpHost("localhost", 9200, "https")};
  
  final SSLContext sslContext = SSLContextBuilder.create()
  .loadTrustMaterial(null, (chains, authType) -> true).build();
  SSLIOSessionStrategy sessionStrategy = new SSLIOSessionStrategy(sslContext, NoopHostnameVerifier.INSTANCE);
  
  final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
  credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("username", "passwowd"));
  
  RestClient restClient = RestClient.builder(hosts)
  .setHttpClientConfigCallback(httpClientBuilder ->
  httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider)
  .setSSLStrategy(sessionStrategy)
  .disableAuthCaching()
  ).setRequestConfigCallback(requestConfigCallback ->
  requestConfigCallback.setConnectTimeout(30000).setSocketTimeout(300000))
  .build();
  
  EasysearchTransport transport = new RestClientTransport(
  restClient, new JacksonJsonpMapper());
  return new EasysearchClient(transport);
  }
  ```
  
  这段代码，简单来说，就是：
  
  1. **连上 Easysearch：** 我们要用 HTTPS 连接到本地的 9200 端口。
  2. **搞定证书：** 这里为了方便，我们信任了所有证书（注意！生产环境一定要配置好你们自己的证书）。
  3. **填上用户名密码：** 这里需要填上你的用户名和密码。
  4. **设置连接参数：** 设置了连接超时时间（30 秒）和读取超时时间（300 秒）。
  5. **创建客户端：** 最后，我们就创建好了一个 `EasysearchClient` 实例，可以开始干活了！
  
  ### 举个栗子：批量操作
  
  下面的例子演示了怎么用 `bulk` API 来批量索引数据：
  
  ```java
  public static void bulk() throws Exception {
  
  String json2 = "{"
  + " \"@timestamp\": \"2023-01-08T22:50:13.059Z\","
  + " \"agent\": {"
  + " \"version\": \"7.3.2\","
  + " \"type\": \"filebeat\","
  + " \"ephemeral_id\": \"3ff1f2c8-1f7f-48c2-b560-4272591b8578\","
  + " \"hostname\": \"ba-0226-msa-fbl-747db69c8d-ngff6\""
  + " }"
  + "}";
  
  EasysearchClient client = create();
  
  BulkRequest.Builder br = new BulkRequest.Builder();
  br.index("test1");
  for (int i = 0; i < 10; i++) {
  BulkOperation.Builder builder = new BulkOperation.Builder();
  IndexOperation.Builder indexBuilder = new IndexOperation.Builder();
  builder.index(indexBuilder.document(JsonData.fromJson(json2)).build());
  br.operations(builder.build());
  }
  
  for (int i = 0; i < 10; i++) {
  BulkOperation.Builder builder = new BulkOperation.Builder();
  IndexOperation.Builder indexBuilder = new IndexOperation.Builder();
  indexBuilder.document(JsonData.fromJson(json2)).index("test2");
  builder.index(indexBuilder.build());
  br.operations(builder.build());
  }
  
  for (int i = 0; i < 10; i++) {
  Map map = new HashMap<>();
  map.put("@timestamp", "2023-01-08T22:50:13.059Z");
  map.put("field1", "value1");
  IndexOperation.Builder indexBuilder = new IndexOperation.Builder();
  indexBuilder.document(map).index("test3");
  br.operations(new BulkOperation(indexBuilder.build()));
  }
  
  
  BulkResponse bulkResponse = client.bulk(br.build());
  if (bulkResponse.errors()) {
  for (BulkResponseItem item : bulkResponse.items()) {
  System.out.println(item.toString());
  }
  }
  client._transport().close();
  
  }
  ```
  
  这个例子里，我们一口气把数据批量索引到了 `test1`、`test2` 和 `test3` 这三个索引里，
  并且展示了三种在 bulk API 中构建 IndexOperation 的方式，虽然它们最终都能实现将文档索引到 Easysearch，但在使用场景和灵活性上还是有一些区别的：
  
  这段代码的核心是利用 BulkRequest.Builder 来构建一个批量请求，并通过 br.operations(...) 方法添加多个操作。而每个操作，在这个例子里，都是一个 IndexOperation，也就是索引一个文档。IndexOperation 可以通过 IndexOperation.Builder 来创建。
  
  三种方式的区别主要体现在如何构建 IndexOperation 里的 document 部分，也就是要索引的文档内容。
  
  **第一种方式：使用 JsonData.fromJson(json2) 且不指定索引。**
  
  特点：
  使用 JsonData.fromJson(json2) 将一个 JSON 字符串直接转换成 JsonData 对象作为文档内容。
  这里没有在 IndexOperation.Builder 上调用 index() 方法来指定索引名称。由于没有在每个 IndexOperation 中指定索引，这个索引名称将回退到 BulkRequest.Builder 上设置的索引，即 br.index("test1")，所以这 10 个文档都会被索引到 test1。
  当你需要将一批相同结构的 JSON 文档索引到同一个索引时，这种方式比较简洁。
  
  **第二种方式：使用 JsonData.fromJson(json2) 并指定索引**
  
  特点：
  同样使用 JsonData.fromJson(json2) 将 JSON 字符串转换成 JsonData 对象。
  关键区别在于，这里在 IndexOperation.Builder 上调用了 index("test2")，为每个操作单独指定了索引名称。
  这 10 个文档会被索引到 test2，即使 BulkRequest.Builder 上设置了 index("test1") 也没用，因为 IndexOperation 里的设置优先级更高。
  当你需要将一批相同结构的 JSON 文档索引到不同的索引时，就需要使用这种方式来分别指定索引。
  
  **第三种方式：使用 Map 并指定索引**
  
  特点：
  使用 Map 来构建文档内容，这种方式更加灵活，可以构建任意结构的文档。
  同样在 IndexOperation.Builder 上调用了 index("test3") 指定了索引名称。
  使用 new BulkOperation(indexBuilder.build()) 代替之前的 builder.index(indexBuilder.build())， 这是等价的。
  这 10 个文档会被索引到 test3。
  当你需要索引的文档结构不固定，或者你需要动态构建文档内容时，使用 Map 是最佳选择。例如，你可以根据不同的业务逻辑，往 Map 里添加不同的字段。
  
  ## 总结
  
  这次 easysearch-client 2.0.x Java 客户端的更新真的很给力，强烈建议大家升级体验！相信我，用了新版客户端，你的开发效率绝对会提升一大截！
  
  ---
  
  **想要了解更多？**
  
  
• 客户端 Maven 地址： [https://mvnrepository.com/arti ... 2.0.2](https://mvnrepository.com/arti ... /2.0.2)
  
  
• 更详细的文档和示例代码在 [官网](https://infinilabs.cn/docs/lat ... t-api/) 持续更新中，请随时关注！
  
  大家有啥问题或者建议，也欢迎随时反馈！
  
  

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

探索 Easysearch 的无限可能，与 INFINI Labs 共赴搜索技术前沿！

随着数字化转型的加速，搜索技术已成为企业数据洞察的核心。INFINI Labs 作为搜索创新技术的引领者，诚邀所有对 Easysearch 搜索引擎感兴趣的开发者、技术爱好者及合作伙伴，参加我们即将于 2024 年 10 月 19 日至 20 日举办的第 3 期 Easysearch 线上免费培训活动。这不仅是一场知识的盛宴，更是技能提升的加速器，将助您在职业道路上迈出坚实的一步！

活动亮点

• 系统课程，全面深入：从 Easysearch 的基本概念到环境搭建，再到高级功能的实战应用，INFINI Labs 的技术专家将为您带来全面而深入的讲解，确保每位参与者都能收获满满。
  
• 实战导向，解决痛点：课程设计紧贴实际需求，旨在帮助学员掌握 Easysearch 的核心技术，有效解决工作中的搜索需求和技术难题，让理论知识迅速转化为实践能力。
  
• 认证证书，助力进阶：后期 INFINI Labs 将推出 Easysearch 认证考试。通过考试的学员将获得官方认证的 Easysearch 证书，为您的职业发展增添强力背书，开启职业生涯的新篇章。
  
  

  培训时间
  

  
  2024 年 10 月 19 日至 20 日（周六、周日）共两天，每天具体培训时间：
  
  

• 上午 09:30 ~ 11:30
  
• 下午 14:00 ~ 16:00
  
  

  培训内容概览
  

  
  

  第一阶段：初识 Easysearch
  

  
  

• Easysearch 环境搭建与对比，了解其与 Elasticsearch 的差异；
  
• 功能初探：身份验证、数据脱敏、权限控制等，全面掌握 Easysearch 的基础功能；
  
• 容灾技术：兼容性验证、跨集群复制等，确保您的搜索服务稳定可靠；
  
  

  第二阶段：Easysearch 高阶使用
  

  
  

• 深度探析：性能压测、数据迁移、请求管理等，提升 Easysearch 的使用效率；
  
• 高级功能：快照管理、可视化看板、告警功能等，让您的搜索服务更加智能；
  
• 生态集成：Filebeat、Logstash、LangChain 等，轻松实现 Easysearch 与其他工具的集成；
  
  

  参与方式
  

  
  本次活动完全免费，名额有限，请尽快报名，同时微信扫码添加小助手进群（培训会议地址将在微信群公布），锁定您的学习席位！
  
  扫码或点击 [我要报名](https://hdxu.cn/Qo9UJ)​
  
  
  
  👉 立即行动：不要错过这次提升自我、与行业精英共成长的宝贵机会。让我们相聚云端，共同探索 Easysearch 的无限可能，开启技术进阶的新篇章！
  
  

  参会提示
  

  
  

• 培训内容涉及动手实践，请务必自备电脑（Windows 系统环境请提前安装好 Linux 虚拟机）；
  
• 请提前在 INFINI Labs 官网下载对应平台最新安装包（INFINI Easysearch、INFINI Gateway、INFINI Console）；
  
• 下载地址：[https://infinilabs.cn/download](https://infinilabs.cn/download)
  
  

  联系我们
  

  
  如有任何疑问，欢迎通过微信添加 [小助手：INFINI-Labs] 与我们联系。
  
  INFINI Labs 期待与您相约，共赴这场技术盛宴！
  
  

  关于 Easysearch
  

  
  
  
  Easysearch 是一个分布式的近实时搜索与分析引擎，基于开源的 Apache Lucene 构建。它旨在提供一个自主可控、轻量级的 Elasticsearch 可替代版本，并不断完善和支持更多的企业级功能。与 Elasticsearch 相比，Easysearch 更注重搜索业务场景的优化，同时保持其产品的简洁与易用性。
  
  详情参见：[Easysearch 介绍](https://infinilabs.cn/products/easysearch)
  
  

  关于极限科技（INFINI Labs）
  

  
  
  
  极限科技，全称极限数据（北京）科技有限公司，是一家专注于实时搜索与数据分析的软件公司。旗下品牌极限实验室（INFINI Labs）致力于打造极致易用的数据探索与分析体验。
  
  极限科技是一支年轻的团队，采用天然分布式的方式来进行远程协作，员工分布在全球各地，希望通过努力成为中国乃至全球企业大数据实时搜索分析产品的首选，为中国技术品牌输出添砖加瓦。
  
  官网：[https://infinilabs.cn](https://infinilabs.cn)