与 INFINI Labs 的技术专家面对面，第一时间了解极限实验室的发布最新产品和功能特性，通过动手实战，快速掌握最前沿的搜索技术，并用于实际项目中。欢迎大家扫描海报二维码免费报名参加。

活动时间：2024-01-18 13:30～17:30

活动地点：北京市海淀区 Wework 辉煌时代大厦 3 楼 3E 会议室

分享议题

• Easysearch 总体介绍及搭建实战
• 基于 INFINI Console 实现跨版本、跨引擎统一管控
• Elasticsearch -> Easysearch 在线迁移实操

参会提示

• 请务必自备电脑（Windows 系统环境请提前安装好 Linux 虚拟机）
• 请提前在 INFINI Labs 官网下载对应平台最新安装包（INFINI Easysearch、INFINI Gateway、INFINI Console）
• 下载地址：https://www.infinilabs.com/download
• 如有任何疑问可添加 INFINI Labs 小助手（微信号: INFINI-Labs）进行联系

极限科技诚招全职搜索运维工程师（Elasticsearch/Easysearch）！

欢迎搜索技术热爱者加入我们，共同打造高效、智能的搜索解决方案！

在招岗位介绍

岗位名称

搜索运维工程师（Elasticsearch/Easysearch）

Base：北京/西安
薪资待遇：10-15K，五险一金，双休等

岗位职责

1. 负责客户现场的 Elasticsearch/Easysearch/OpenSearch 搜索引擎集群的日常维护、监控和优化，确保集群的高可用性和性能稳定;
2. 协助客户进行搜索引擎集群的部署、配置及版本升级;
3. 排查和解决 Elasticsearch/Easysearch/OpenSearch 集群中的各种技术问题，及时响应并处理集群异常;
4. 根据业务需求设计和实施搜索索引的调优、数据迁移和扩展方案;
5. 负责与客户沟通，提供技术支持及相关培训，确保客户需求得到有效满足;
6. 制定并实施搜索引擎的备份、恢复和安全策略，保障数据安全;
7. 与内部研发团队和外部客户进行协作，推动集群性能改进和功能优化。

岗位要求

1. 全日制本科及以上学历，2 年以上运维工作经验;
2. 拥有 Elasticsearch/Easysearch/OpenSearch 使用经验，熟悉搜索引擎的原理、架构和相关生态工具（如 Logstash、Kibana 等）;
3. 熟悉 Linux 操作系统的使用及常见性能调优方法;
4. 熟练掌握 Shell 或 Python 等至少一种脚本语言，能够编写自动化运维脚本;
5. 具有优秀的问题分析与解决能力，能够快速应对突发情况;
6. 具备良好的沟通能力和团队合作精神，能够接受客户驻场工作;
7. 提供 五险一金，享有带薪年假及法定节假日。

加分项

1. 计算机科学、信息技术或相关专业;
2. 具备丰富的大规模分布式系统运维经验;
3. 熟悉 Elasticsearch/Easysearch/OpenSearch 分片、路由、查询优化等高级功能;
4. 拥有 Elastic Certified Engineer 认证;
5. 具备大规模搜索引擎集群设计、扩展和调优经验;
6. 熟悉其他搜索引擎技术（如 Solr、Lucene）者优先;
7. 熟悉大数据处理相关技术（比如: Kafka 、Flink 等）者优先。

简历投递

1. 邮件：hello@infini.ltd（邮件标题请备注姓名+求职岗位）
2. 微信：INFINI-Labs （加微请备注求职岗位）

关于极限科技（INFINI Labs）

极限科技，全称极限数据（北京）科技有限公司，是一家专注于实时搜索与数据分析的软件公司。旗下品牌极限实验室（INFINI Labs）致力于打造极致易用的数据探索与分析体验。

极限科技是一支年轻的团队，采用天然分布式的方式来进行远程协作，员工分布在全球各地，希望通过努力成为中国乃至全球企业大数据实时搜索分析产品的首选，为中国技术品牌输出添砖加瓦。

官网：https://infinilabs.cn

我们在做什么

极限科技（INFINI Labs）正在致力于以下几个核心方向：

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

2、打造下一代实时搜索引擎 INFINI Pizza
INFINI Pizza 是一个分布式混合搜索数据库系统。我们的使命是充分利用现代硬件和人工智能的潜力，为企业提供量身定制的实时智能搜索体验。我们致力于满足具有挑战性的环境中高并发和高吞吐量的需求，同时提供无缝高效的搜索功能。详情参见：https://pizza.rs

3、为现代团队打造的统一搜索与 AI 智能助手 Coco AI
Coco AI 是一款完全开源的统一搜索与 AI 助手平台，它通过统一搜索入口，连接企业内外部的异构数据源，融合大模型能力，帮助团队高效访问知识，智能决策协作。详情参见：https://coco.rs

4、积极参与全球开源生态建设
通过开源 Coco AI、Console、Gateway、Agent、Loadgen 等搜索领域产品和社区贡献，推动全球开源技术的发展，提升中国在全球开源领域的影响力。INFINI Labs Github 主页：https://github.com/infinilabs

5、提供专业服务
为客户提供包括搜索技术支持、迁移服务、定制解决方案和培训在内的全方位服务。

6、提供国产化搜索解决方案
针对中国市场的特殊需求，提供符合国产化标准的搜索产品和解决方案，帮助客户解决使用 Elasticsearch 时遇到的挑战。

极限科技（INFINI Labs）通过这些努力，旨在成为全球领先的实时搜索和数据分析解决方案提供商。

背景：bulk 优化的应用

在 ES 的写入优化里，bulk 操作被广泛地用于批量处理数据。bulk 操作允许用户一次提交多个数据操作，如索引、更新、删除等，从而提高数据处理效率。bulk 操作的实现原理是，将数据操作请求打包成 HTTP 请求，并批量提交给 Elasticsearch 服务器。这样，Elasticsearch 服务器就可以一次处理多个数据操作，从而提高处理效率。

这种优化的核心价值在于减少了网络往返的次数和连接建立的开销。每一次单独的写入操作都需要经历完整的请求-响应周期，而批量写入则是将多个操作打包在一起，用一次通信完成原本需要多次交互的工作。这不仅仅节省了时间，更重要的是释放了系统资源，让服务器能够专注于真正的数据处理，而不是频繁的协议握手和状态维护。

这样的批量请求的确是可以优化写入请求的效率，让 ES 集群获得更多的资源去做写入请求的集中处理。但是除了客户端与 ES 集群的通讯效率优化，还有其他中间过程能优化么？

Gateway 的优化点

bulk 的优化理念是将日常零散的写入需求做集中化的处理，尽量减低日常请求的损耗，完成资源最大化的利用。简而言之就是“好钢用在刀刃上”。

但是 ES 在收到 bulk 写入请求后，也是需要协调节点根据文档的 id 计算所属的分片来将数据分发到对应的数据节点的。这个过程也是有一定损耗的，如果 bulk 请求中数据分布的很散，每个分片都需要进行写入，原本 bulk 集中写入的需求优势则还是没有得到最理想化的提升。

gateway 的写入加速则对 bulk 的优化理念的最大化补全。

gateway 可以本地计算每个索引文档对应后端 Elasticsearch 集群的目标存放位置，从而能够精准的进行写入请求定位。

在一批 bulk 请求中，可能存在多个后端节点的数据，bulk_reshuffle 过滤器用来将正常的 bulk 请求打散，按照目标节点或者分片进行拆分重新组装，避免 Elasticsearch 节点收到请求之后再次进行请求分发， 从而降低 Elasticsearch 集群间的流量和负载，也能避免单个节点成为热点瓶颈，确保各个数据节点的处理均衡，从而提升集群总体的索引吞吐能力。

整理的优化思路如下图：

优化实践

那我们来实践一下，看看 gateway 能提升多少的写入。

这里我们分 2 个测试场景：

1. 基础集中写入测试，不带文档 id，直接批量写入。这个场景更像是日志或者监控数据采集的场景。
2. 带文档 id 的写入测试，更偏向搜索场景或者大数据批同步的场景。

2 个场景都进行直接写入 ES 和 gateway 转发 ES 的效率比对。

测试材料除了需要备一个网关和一套 es 外，其余的内容如下：

测试索引 mapping 一致，名称区分：

PUT gateway_bulk_test
{
  "settings": {
    "number_of_shards": 6,
    "number_of_replicas": 0
  },
  "mappings": {
    "properties": {
      "timestamp": {
        "type": "date",
        "format": "strict_date_optional_time"
      },
      "field1": {
        "type": "keyword"
      },
      "field2": {
        "type": "keyword"
      },
      "field3": {
        "type": "keyword"
      },
      "field4": {
        "type": "integer"
      },
      "field5": {
        "type": "keyword"
      },
      "field6": {
        "type": "float"
      }
    }
  }
}

PUT bulk_test
{
  "settings": {
    "number_of_shards": 6,
    "number_of_replicas": 0
  },
  "mappings": {
    "properties": {
      "timestamp": {
        "type": "date",
        "format": "strict_date_optional_time"
      },
      "field1": {
        "type": "keyword"
      },
      "field2": {
        "type": "keyword"
      },
      "field3": {
        "type": "keyword"
      },
      "field4": {
        "type": "integer"
      },
      "field5": {
        "type": "keyword"
      },
      "field6": {
        "type": "float"
      }
    }
  }
}

gateway 的配置文件如下：

path.data: data
path.logs: log

entry:
  - name: my_es_entry
    enabled: true
    router: my_router
    max_concurrency: 200000
    network:
      binding: 0.0.0.0:8000

flow:
  - name: async_bulk
    filter:
      - bulk_reshuffle:
          when:
            contains:
              _ctx.request.path: /_bulk
          elasticsearch: prod
          level: node
          partition_size: 1
          fix_null_id: true
      - elasticsearch:
          elasticsearch: prod #elasticsearch configure reference name
          max_connection_per_node: 1000 #max tcp connection to upstream, default for all nodes
          max_response_size: -1 #default for all nodes
          balancer: weight
          refresh: # refresh upstream nodes list, need to enable this feature to use elasticsearch nodes auto discovery
            enabled: true
            interval: 60s
          filter:
            roles:
              exclude:
                - master

router:
  - name: my_router
    default_flow: async_bulk

elasticsearch:
  - name: prod
    enabled: true
    endpoints:
      - https://127.0.0.1:9221
      - https://127.0.0.1:9222
      - https://127.0.0.1:9223
    basic_auth:
      username: admin
      password: admin

pipeline:
  - name: bulk_request_ingest
    auto_start: true
    keep_running: true
    retry_delay_in_ms: 1000
    processor:
      - bulk_indexing:
          max_connection_per_node: 100
          num_of_slices: 3
          max_worker_size: 30
          idle_timeout_in_seconds: 10
          bulk:
            compress: false
            batch_size_in_mb: 10
            batch_size_in_docs: 10000
          consumer:
            fetch_max_messages: 100
          queue_selector:
            labels:
              type: bulk_reshuffle

测试脚本如下：

#!/usr/bin/env python3
"""
ES Bulk写入性能测试脚本

"""

import hashlib
import json
import random
import string
import time
from typing import List, Dict, Any

import requests
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
import urllib3

# 禁用SSL警告
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class ESBulkTester:
    def __init__(self):
        # 配置变量 - 可修改
        self.es_configs = [
            {
                "name": "ES直连",
                "url": "https://127.0.0.1:9221",
                "index": "bulk_test",
                "username": "admin",  # 修改为实际用户名
                "password": "admin",  # 修改为实际密码
                "verify_ssl": False  # HTTPS需要SSL验证
            },
            {
                "name": "Gateway代理",
                "url": "http://localhost:8000",
                "index": "gateway_bulk_test",
                "username": None,  # 无需认证
                "password": None,
                "verify_ssl": False
            }
        ]
        self.batch_size = 10000  # 每次bulk写入条数
        self.log_interval = 100000  # 每多少次bulk写入输出日志

        # ID生成规则配置 - 前2位后5位
        self.id_prefix_start = 1
        self.id_prefix_end = 999      # 前3位: 01-999
        self.id_suffix_start = 1
        self.id_suffix_end = 9999   # 后4位: 0001-9999

        # 当前ID计数器
        self.current_prefix = self.id_prefix_start
        self.current_suffix = self.id_suffix_start

    def generate_id(self) -> str:
        """生成固定规则的ID - 前2位后5位"""
        id_str = f"{self.current_prefix:02d}{self.current_suffix:05d}"

        # 更新计数器
        self.current_suffix += 1
        if self.current_suffix > self.id_suffix_end:
            self.current_suffix = self.id_suffix_start
            self.current_prefix += 1
            if self.current_prefix > self.id_prefix_end:
                self.current_prefix = self.id_prefix_start

        return id_str

    def generate_random_hash(self, length: int = 32) -> str:
        """生成随机hash值"""
        random_string = ''.join(random.choices(string.ascii_letters + string.digits, k=length))
        return hashlib.md5(random_string.encode()).hexdigest()

    def generate_document(self) -> Dict[str, Any]:
        """生成随机文档内容"""
        return {
            "timestamp": datetime.now().isoformat(),
            "field1": self.generate_random_hash(),
            "field2": self.generate_random_hash(),
            "field3": self.generate_random_hash(),
            "field4": random.randint(1, 1000),
            "field5": random.choice(["A", "B", "C", "D"]),
            "field6": random.uniform(0.1, 100.0)
        }

    def create_bulk_payload(self, index_name: str) -> str:
        """创建bulk写入payload"""
        bulk_data = []

        for _ in range(self.batch_size):
            #doc_id = self.generate_id()
            doc = self.generate_document()

            # 添加index操作
            bulk_data.append(json.dumps({
                "index": {
                    "_index": index_name,
            #        "_id": doc_id
                }
            }))
            bulk_data.append(json.dumps(doc))

        return "\n".join(bulk_data) + "\n"

    def bulk_index(self, config: Dict[str, Any], payload: str) -> bool:
        """执行bulk写入"""
        url = f"{config['url']}/_bulk"
        headers = {
            "Content-Type": "application/x-ndjson"
        }

        # 设置认证信息
        auth = None
        if config.get('username') and config.get('password'):
            auth = (config['username'], config['password'])

        try:
            response = requests.post(
                url,
                data=payload,
                headers=headers,
                auth=auth,
                verify=config.get('verify_ssl', True),
                timeout=30
            )
            return response.status_code == 200
        except Exception as e:
            print(f"Bulk写入失败: {e}")
            return False

    def refresh_index(self, config: Dict[str, Any]) -> bool:
        """刷新索引"""
        url = f"{config['url']}/{config['index']}/_refresh"

        # 设置认证信息
        auth = None
        if config.get('username') and config.get('password'):
            auth = (config['username'], config['password'])

        try:
            response = requests.post(
                url,
                auth=auth,
                verify=config.get('verify_ssl', True),
                timeout=10
            )
            success = response.status_code == 200
            print(f"索引刷新{'成功' if success else '失败'}: {config['index']}")
            return success
        except Exception as e:
            print(f"索引刷新失败: {e}")
            return False

    def run_test(self, config: Dict[str, Any], round_num: int, total_iterations: int = 100000):
        """运行性能测试"""
        test_name = f"{config['name']}-第{round_num}轮"
        print(f"\n开始测试: {test_name}")
        print(f"ES地址: {config['url']}")
        print(f"索引名称: {config['index']}")
        print(f"认证: {'是' if config.get('username') else '否'}")
        print(f"每次bulk写入: {self.batch_size}条")
        print(f"总计划写入: {total_iterations * self.batch_size}条")
        print("-" * 50)

        start_time = time.time()
        success_count = 0
        error_count = 0

        for i in range(1, total_iterations + 1):
            payload = self.create_bulk_payload(config['index'])

            if self.bulk_index(config, payload):
                success_count += 1
            else:
                error_count += 1

            # 每N次输出日志
            if i % self.log_interval == 0:
                elapsed_time = time.time() - start_time
                rate = i / elapsed_time if elapsed_time > 0 else 0
                print(f"已完成 {i:,} 次bulk写入, 耗时: {elapsed_time:.2f}秒, 速率: {rate:.2f} bulk/秒")

        end_time = time.time()
        total_time = end_time - start_time
        total_docs = total_iterations * self.batch_size

        print(f"\n{test_name} 测试完成!")
        print(f"总耗时: {total_time:.2f}秒")
        print(f"成功bulk写入: {success_count:,}次")
        print(f"失败bulk写入: {error_count:,}次")
        print(f"总文档数: {total_docs:,}条")
        print(f"平均速率: {success_count/total_time:.2f} bulk/秒")
        print(f"文档写入速率: {total_docs/total_time:.2f} docs/秒")
        print("=" * 60)

        return {
            "test_name": test_name,
            "config_name": config['name'],
            "round": round_num,
            "es_url": config['url'],
            "index": config['index'],
            "total_time": total_time,
            "success_count": success_count,
            "error_count": error_count,
            "total_docs": total_docs,
            "bulk_rate": success_count/total_time,
            "doc_rate": total_docs/total_time
        }

    def run_comparison_test(self, total_iterations: int = 10000):
        """运行双地址对比测试"""
        print("ES Bulk写入性能测试开始")
        print(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        print("=" * 60)

        results = []
        rounds = 2  # 每个地址测试2轮

        # 循环测试所有配置
        for config in self.es_configs:
            print(f"\n开始测试配置: {config['name']}")
            print("*" * 40)

            for round_num in range(1, rounds + 1):
                # 运行测试
                result = self.run_test(config, round_num, total_iterations)
                results.append(result)

                # 每轮结束后刷新索引
                print(f"\n第{round_num}轮测试完成，执行索引刷新...")
                self.refresh_index(config)

                # 重置ID计数器
                if round_num == 1:
                    # 第1轮：使用初始ID范围（新增数据）
                    print("第1轮：新增数据模式")
                else:
                    # 第2轮：重复使用相同ID（更新数据模式）
                    print("第2轮：数据更新模式，复用第1轮ID")
                    self.current_prefix = self.id_prefix_start
                    self.current_suffix = self.id_suffix_start

                print(f"{config['name']} 第{round_num}轮测试结束\n")

        # 输出对比结果
        print("\n性能对比结果:")
        print("=" * 80)

        # 按配置分组显示结果
        config_results = {}
        for result in results:
            config_name = result['config_name']
            if config_name not in config_results:
                config_results[config_name] = []
            config_results[config_name].append(result)

        for config_name, rounds_data in config_results.items():
            print(f"\n{config_name}:")
            total_time = 0
            total_bulk_rate = 0
            total_doc_rate = 0

            for round_data in rounds_data:
                print(f"  第{round_data['round']}轮:")
                print(f"    耗时: {round_data['total_time']:.2f}秒")
                print(f"    Bulk速率: {round_data['bulk_rate']:.2f} bulk/秒")
                print(f"    文档速率: {round_data['doc_rate']:.2f} docs/秒")
                print(f"    成功率: {round_data['success_count']/(round_data['success_count']+round_data['error_count'])*100:.2f}%")

                total_time += round_data['total_time']
                total_bulk_rate += round_data['bulk_rate']
                total_doc_rate += round_data['doc_rate']

            avg_bulk_rate = total_bulk_rate / len(rounds_data)
            avg_doc_rate = total_doc_rate / len(rounds_data)

            print(f"  平均性能:")
            print(f"    总耗时: {total_time:.2f}秒")
            print(f"    平均Bulk速率: {avg_bulk_rate:.2f} bulk/秒")
            print(f"    平均文档速率: {avg_doc_rate:.2f} docs/秒")

        # 整体对比
        if len(config_results) >= 2:
            config_names = list(config_results.keys())
            config1_avg = sum([r['bulk_rate'] for r in config_results[config_names[0]]]) / len(config_results[config_names[0]])
            config2_avg = sum([r['bulk_rate'] for r in config_results[config_names[1]]]) / len(config_results[config_names[1]])

            if config1_avg > config2_avg:
                faster = config_names[0]
                rate_diff = config1_avg - config2_avg
            else:
                faster = config_names[1]
                rate_diff = config2_avg - config1_avg

            print(f"\n整体性能对比:")
            print(f"{faster} 平均性能更好，bulk速率高 {rate_diff:.2f} bulk/秒")
            print(f"性能提升: {(rate_diff/min(config1_avg, config2_avg)*100):.1f}%")

def main():
    """主函数"""
    tester = ESBulkTester()

    # 运行测试（每次bulk 1万条，300次bulk = 300万条文档）
    tester.run_comparison_test(total_iterations=300)

if __name__ == "__main__":
    main()

1. 日志场景：不带 id 写入

测试条件：

1. bulk 写入数据不带文档 id
2. 每批次 bulk 10000 条数据，总共写入 30w 数据

这里把

反馈结果：

性能对比结果:
================================================================================

ES直连:
  第1轮:
    耗时: 152.29秒
    Bulk速率: 1.97 bulk/秒
    文档速率: 19699.59 docs/秒
    成功率: 100.00%
  平均性能:
    总耗时: 152.29秒
    平均Bulk速率: 1.97 bulk/秒
    平均文档速率: 19699.59 docs/秒

Gateway代理:
  第1轮:
    耗时: 115.63秒
    Bulk速率: 2.59 bulk/秒
    文档速率: 25944.35 docs/秒
    成功率: 100.00%
  平均性能:
    总耗时: 115.63秒
    平均Bulk速率: 2.59 bulk/秒
    平均文档速率: 25944.35 docs/秒

整体性能对比:
Gateway代理 平均性能更好，bulk速率高 0.62 bulk/秒
性能提升: 31.7%

2. 业务场景：带文档 id 的写入

测试条件：

1. bulk 写入数据带有文档 id，两次测试写入的文档 id 生成规则一致且重复。
2. 每批次 bulk 10000 条数据，总共写入 30w 数据

这里把 py 脚本中 第 99 行 和 第 107 行的注释打开。

反馈结果：

性能对比结果:
================================================================================

ES直连:
  第1轮:
    耗时: 155.30秒
    Bulk速率: 1.93 bulk/秒
    文档速率: 19317.39 docs/秒
    成功率: 100.00%
  平均性能:
    总耗时: 155.30秒
    平均Bulk速率: 1.93 bulk/秒
    平均文档速率: 19317.39 docs/秒

Gateway代理:
  第1轮:
    耗时: 116.73秒
    Bulk速率: 2.57 bulk/秒
    文档速率: 25700.06 docs/秒
    成功率: 100.00%
  平均性能:
    总耗时: 116.73秒
    平均Bulk速率: 2.57 bulk/秒
    平均文档速率: 25700.06 docs/秒

整体性能对比:
Gateway代理 平均性能更好，bulk速率高 0.64 bulk/秒
性能提升: 33.0%

小结

不管是日志场景还是业务价值更重要的大数据或者搜索数据同步场景， gateway 的写入加速都能平稳的节省 25%-30% 的写入耗时。

关于极限网关（INFINI Gateway）

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

官网文档：https://docs.infinilabs.com/gateway
开源地址：https://github.com/infinilabs/gateway

作者：金多安，极限科技（INFINI Labs）搜索运维专家，Elastic 认证专家，搜索客社区日报责任编辑。一直从事与搜索运维相关的工作，日常会去挖掘 ES / Lucene 方向的搜索技术原理，保持搜索相关技术发展的关注。
原文：https://infinilabs.cn/blog/2025/gateway-bulk-write-performance-optimization/

本次活动由 搜索客社区、极限科技（INFINI Labs） 联合举办，活动邀请到 INFINI Labs 搜索运维专家 金端 来分享 Elasticsearch 中一些典型的“双刃剑”特性，以及如何在实际使用中权衡和应对。欢迎预约直播观看 ~

活动主题：Elasticsearch 的一些“双刃剑”特性
活动时间：2025 年 03 月 26 日 19:00-20:00（周三）
活动形式：微信视频号（极限实验室）直播
报名方式：关注或扫码海报中的二维码进行预约

嘉宾介绍

金端，极限科技（INFINI Labs）搜索运维专家，Elasticsearch 官方认证工程师，搜索客社区日报编辑。在 ES 运维使用方面具有丰富的实战经验。对 ES/lucene 搜索分析方向保持学习和关注。

主题摘要

Elasticsearch 的许多特性确实是一把“双刃剑”，它们在提供强大功能的同时，也带来了潜在的使用瓶颈和运维挑战。分享主要讨论一些典型的“双刃剑”特性，以及如何在实际使用中权衡和应对。

参与有奖

本次直播活动将设有福袋抽奖环节，参与就有机会获得 INFINI Labs 周边纪念品，包括 T 恤、鸭舌帽、咖啡杯、指甲刀套件、精品围巾等等（图片仅供参考，款式、颜色与尺码随机）。

活动交流

本活动设有 Meetup 技术交流群，可添加小助手微信拉群，与更多小伙伴一起学习交流。

Meetup 讲师招募

搜索客社区 Meetup 的成功举办，离不开社区小伙伴的热情参与。目前社区讲师招募计划也在持续进行中，我们诚挚邀请各位技术大咖、行业精英踊跃提交演讲议题，与大家分享您的经验。

讲师报名链接：http://cfp.searchkit.cn
或扫描下方二维码，立刻报名成为讲师！

Meetup 活动聚焦 AI 与搜索领域的最新动态，以及数据实时搜索分析、向量检索、技术实践与案例分析、日志分析、安全等领域的深度探讨。

我们热切期待您的精彩分享！

往期回顾

• 【第 7 期】搜索客 Meetup | 开源智能搜索与知识库管理，极限科技 Coco AI 产品介绍
• 【第 6 期】搜索客 Meetup | Easysearch 请求限速漫谈
• 【第 5 期】搜索客 Meetup | 最强开源 Elasticsearch 多集群管理工具 INFINI Console 动手实战
• 【第 4 期】搜索客 Meetup | INFINI Pizza 网站 SVG 动画这么炫，我教你啊！
• 【第 3 期】搜索客 Meetup | Elasticsearch 的代码结构和写入查询流程的解读 - 下篇
• 【第 2 期】搜索客 Meetup | Elasticsearch 的代码结构和写入查询流程的解读 - 上篇
• 【第 1 期】搜索客 Meetup | Easysearch 结合大模型实现 RAG

关于 搜索客（SearchKit）社区

搜索客社区由 Elasticsearch 中文社区进行全新的品牌升级，以新的 Slogan：“搜索人自己的社区” 为宣言。汇集搜索领域最新动态、精选干货文章、精华讨论、文档资料、翻译与版本发布等，为广大搜索领域从业者提供更为丰富便捷的学习和交流平台。

社区官网：https://searchkit.cn 。

作者：京东零售 崔健

0.前言

• 系统概要：BIP采购系统用于京东采销部门向供应商采购商品，并且提供了多种创建采购单的方式以及采购单审批、回告、下传回传等业务功能

• 系统价值：向供应商采购商品增加库存，满足库存周转及客户订单的销售，供应链最重要的第一环节

1.背景

采购系统在经历了多年的迭代后，在数据库查询层面面临巨大的性能挑战。核心根因主要有以下几方面：

• 复杂查询多，历史上通过MySQL和JED承载了过多的检索过滤条件，时至今日很难推动接口使用方改变调用方式

• 数据量大，随着业务的持续发展，带来了海量的数据增长（日均150万单左右，订单主表/子表/渠道表/扩展表分别都是：6.5亿行，订单明细表/分配表：9.2亿行，日志表：60亿行）

• 数据模型复杂，订单完整数据分布在20+张表，经常需要多表join

引入的主要问题有：

• 业务层面：

  • 订单列表页查询/导出体验差，性能非常依赖输入条件，尤其是在面对订单数据倾斜的时候，部分用户无法查询/导出超过半个月以上的订单

  • 查询条件不合理，1.归档筛选条件，技术词汇透传到业务，导致相同周期的单子无法一键查询/导出，需要切换“是否归档”查询全部；2.无法区分“需要仓库收货”类的单子，大部分业务同事主要关注这类单子的履约情况

• 技术层面：

  • 慢SQL多，各种多表关联复杂条件查询导致，索引、SQL已经优化道了瓶颈，经常出现数据库负载被拉高

  • 大表多，难在数据库上做DDL，可能会引起核心写库负载升高、主从延迟等问题

  • 模型复杂，开发、迭代成本高，查询索引字段散落在多个表中，导致查询性能下降

2.目标

业务层面：提升核心查询/导出体验，加强查询性能，优化不合理的查询条件

技术层面：1.减少慢SQL，降低数据库负载，提高系统稳定性；2.降低单表数据量级；3.简化数据模型

3.挑战

提升海量数据、复杂场景下的查询性能！

• 采购订单系统 VS C端销售订单系统复杂度对比：

4.方案

思路

优化前



优化后

4.1 降低查询数据量

4.1.1 前期调研

基于历史数据、业务调研分析，采购订单只有8%的订单属于“需要实际送货至京东库房”的范围，也就是拥有完整订单履约流程、业务核心关注时效的。其余订单属于通过客户订单驱动，在采购系统的生命周期只有创建记录



基于以上结论，在与产品达成共识后，提出新的业务领域概念：“入库订单” ，在查询时单独异构这部分订单数据（前期也曾考虑过，直接从写入层面拆分入库订单，但是因为开发成本、改动范围被pass了）。异构这部分数据实际也参考了操作系统、中间件的核心优化思路，缓存访问频次高的热数据

4.1.2 入库订单异构

执行流程





• “入库”订单数据打标

  • 增量订单在创建订单完成时写入；存量订单通过离线表推数

  • 需要订单创建模块先完成改造上线，再同步历史，保证数据不丢

  • 如果在【数据解析模块】处理binlog时无法及时从JimKV获取到订单标识，会补偿反查数据库并回写JimKV，提升其他表的binlog处理效率

• binlog监听

  • 基于公司的【数据订阅】任务，通过消费JMQ实现。其中订阅任务基于订单号进行MQ数据分区，并且在消费端配置不允许消息重试，防止消息时序错乱

  • 其中，根据订单号进行各个表的MQ数据分区，第一版设计可能会引起热分区，导致消费速率变慢，基于这个问题识别到热分区主要是由于频繁更新订单明细数据导致（订单(1)->明细(N)），于是将明细相关表基于自身id进行分区，其他订单纬度表还是基于订单号。这样既不影响订单数据更新的先后顺序，也不会造成热分区、还可以保证明细数据的准确性

• 数据同步

  • 增量数据同步可以采用源库的增量binlog进行解析即可，存量数据通过申请新库/表，进行DTS的存量+增量同步写入，完成binlog生产

  • 以上是在上线前的临时链路，上线后需要切换到源库同步binlog的增量订阅任务，此时依赖“位点回拨”+“数据可重入”。位点回拨基于订阅任务的binlog时间戳，数据可重入依赖上文提到的MQ消费有序以及SQL覆盖写

• 数据校对

  • 以表为纬度，优先统计总数，再随机抽样明细进行比对

  • 目前入库订单量为稳定在5000万，全部实时订单量级6.5亿，降低92%

4.2 提升复杂查询能力

4.2.1 数据准备

• 考虑到异构“入库”订单到JED，虽然数据查询时效性可以有一定保障，但是在复杂查询能力以及识别“非入库”订单还没有支持

• 其中，“非入库”订单业务对于订单数据时效性要求并不高（1.订单创建源于客户订单；2.没有履约流程；3.无需手动操作订单关键节点）

• 所以，考虑将这部分查询能力转移到ES上

ES数据异构过程





• 首先，同步到ES的数据的由“实时+归档”订单组成，其中合计20亿订单，顺带优化了先前归档ES大索引（所有订单放在同一个索引）的问题，改成基于“月份”存储订单，之所以改成月份是因为根据条件查询分两种：1.一定会有查询时间范围（最多3个月）；2.指定单号查询，这种会优先检索单号对应的订单创建时间，再路由到指定索引

• 其次，简化了归档程序流程，历史方案是程序中直接写入【归档JED+归档ES】，现在优化成只写入JED，ES数据通过【数据解析模块】完成，简化归档程序的同时，提高了归档能力的时效性

• 再次，因为ES是存储全量订单，需要支持复杂条件的查询，所以在订单没有物理删除的前提下，【数据解析模块】会过滤所有delete语句，保证全量订单数据的完整性

• 接着，为了提升同步到ES数据的吞吐，在MQ消费端，主要做了两方面优化：1.会根据表和具体操作进行binlog的请求合并；2.降低对于ES内部refresh机制的依赖，将2分钟内更新到ES的数据缓存到JimKV，更新前从缓存中获取

• 最后，上文提到，同步到入库JED，有的表是根据订单号，有的表是根据自身id。那么ES这里，因为NoSQL的设计，和线程并发的问题，为了防止数据错误，只能将所有表数据根据单号路由到相同的MQ分区

4.2.2 查询调度策略设计

优化前，所有的查询请求都会直接落到数据库进行查询，可以高效查询完全取决于用户的筛选条件是否可以精准缩小数据查询范围

优化后，新增动态路由层

• 离线计算T-1的采销/供应商的订单数据倾斜，将数据倾斜情况推送到JimDB集群

• 根据登陆用户、数据延迟要求、查询数据范围，自动调度查询的数据集群，实现高性能的查询请求

查询调度



5.ES主备机制&数据监控

1.主/备ES可以通过DUCC开关，实现动态切换，提升数据可靠性

2.结合公司的业务监控，完成订单数据延迟监控（数据同步模块写入时间-订单创建时间）

3.开启消息队列积压告警

5.1 ES集群主/备机制

1：1ES集群进行互备，应急预案快速切换，保证高可用



5.2 数据监控



6.灰度上线

• 第一步，优先上线数据模块，耗费较多时间的原因：1.整体数据量级以及历史数据复杂度的问题；2.数据同步链路比较长，中间环节多

• 第二步，预发环境验证，流量回放并没有做到长周期的完全自动化，根因：1.项目周期相对紧张；2.新老集群的数据还是有一些区别，回放脚本不够完善

• 第三步，用户功能灰度，主要是借助JDOS的负载均衡策略结合用户erp完成

• 第四部，对外接口灰度，通过控制新代码灰度容器个数，逐步放量



7.成果

平稳切换，无线上问题

8.待办

• 主动监控层面，新增按照天纬度进行数据比对、异常告警的能力，提高问题发现率

• 优化数据模型，对历史无用订单表进行精简，降低开发、运维成本，提升需求迭代效率

• 精简存储集群

  • 逐步下线其他非核心业务存储集群，减少外部依赖，提高系统容错度

  • 目前全量订单ES集群已经可以支持多场景的外部查询，未来考虑是否可以逐步下线入库订单JED

• 识别数据库隐患，基于慢日志监控，重新梳理引入模块，逐步优化，持续降低数据库负载

• MySQL减负，探索其他优化方案，减少数据量存储，提升数据灵活性。优先从业务层面出发，识别库里进行中的僵尸订单的根因，进行分类，强制结束

• 降级方案，当数据同步或者数据库存在异常时，可以做到秒级无感切换，提升业务可用率

9.写在最后

• 为什么没考虑Doris？因为ES是团队应用相对成熟的中间件，处于学习、开发成本考虑

• 未来入库的JED相关表是否可以下掉，用ES完全替代？目前看可以，当初设计冗余入库JED也是出于对于ES不确定性以及数据延迟的角度考虑，而且历史的一部分查询就落在了异构的全量实时订单JED上。现在，JED官方也不是很推荐非route key的查询。最后，现阶段因为降低了数据量和拆分了业务场景，入库JED的查询性能还是非常不错的

• 因为项目排期、个人能力的因素，在方案设计上会有考虑不周的场景，本期只是优化了最核心的业务、技术痛点，未来还有很大持续优化的空间。中间件的使用并不是可以优化数据库性能的银弹，最核心的还是要结合业务以及系统历史背景，在不断纠结当中寻找balance

我们是国内搜索型数据库产品厂商第一梯队的杰出代表，随着业务的快速发展，现开放岗位：搜索运维工程师（ Elasticsearch/Easysearch ），如果有兴趣，请直接拉到文末，扫描二维码或将简历投递至 hello@infini.ltd。

如果您还不了解 极限科技（INFINI Labs）是谁，在做什么，需要什么样的小伙伴，那么请看下文：

我们是谁

极限科技，全称极限数据（北京）科技有限公司，是一家专注于实时搜索与数据分析的软件公司。旗下品牌极限实验室（INFINI Labs）致力于打造极致易用的数据探索与分析体验。

极限科技是一支年轻的团队，采用天然分布式的方式来进行远程协作，员工分布在全球各地，希望通过努力成为中国乃至全球企业大数据实时搜索分析产品的首选，为中国技术品牌输出添砖加瓦。

官网：https://infinilabs.cn

我们在做什么

极限科技（INFINI Labs）正在致力于以下几个核心方向：

1、开发近实时搜索引擎 INFINI Easysearch

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

2、打造下一代实时搜索引擎 INFINI Pizza

INFINI Pizza 是一个分布式混合搜索数据库系统。我们的使命是充分利用现代硬件和人工智能的潜力，为企业提供量身定制的实时智能搜索体验。我们致力于满足具有挑战性的环境中高并发和高吞吐量的需求，同时提供无缝高效的搜索功能。详情参见：https://pizza.rs

3、积极参与全球开源生态建设

通过开源项目和社区贡献，推动全球开源技术的发展，提升中国在全球开源领域的影响力。

4、提供专业服务

为客户提供包括搜索技术支持、迁移服务、定制解决方案和培训在内的全方位服务。

5、提供国产化搜索解决方案

针对中国市场的特殊需求，提供符合国产化标准的搜索产品和解决方案，帮助客户解决使用 Elasticsearch 时遇到的挑战。

极限科技（INFINI Labs）通过这些努力，旨在成为全球领先的实时搜索和数据分析解决方案提供商。

我们期待有才华、有激情的你加入我们，一起探索数据搜索的未来，共同创造无限可能！

在招岗位介绍

岗位名称

搜索运维工程师（Elasticsearch/Easysearch）

岗位职责

1. 负责客户现场的 Elasticsearch/Easysearch/OpenSearch 搜索引擎集群的日常维护、监控和优化，确保集群的高可用性和性能稳定；
2. 协助客户进行搜索引擎集群的部署、配置及版本升级；
3. 排查和解决 Elasticsearch/Easysearch/OpenSearch 集群中的各种技术问题，及时响应并处理集群异常；
4. 根据业务需求设计和实施搜索索引的调优、数据迁移和扩展方案；
5. 负责与客户沟通，提供技术支持及相关培训，确保客户需求得到有效满足；
6. 制定并实施搜索引擎的备份、恢复和安全策略，保障数据安全；
7. 与内部研发团队和外部客户进行协作，推动集群性能改进和功能优化。

岗位要求

1. 全日制本科及以上学历，5 年以上运维工作经验；
2. 拥有 Elasticsearch/Easysearch/OpenSearch 使用经验，熟悉搜索引擎的原理、架构和相关生态工具（如 Logstash、Kibana 等）；
3. 熟悉 Linux 操作系统的使用及常见性能调优方法；
4. 熟练掌握 Shell 或 Python 等至少一种脚本语言，能够编写自动化运维脚本；
5. 具有优秀的问题分析与解决能力，能够快速应对突发情况；
6. 具备良好的沟通能力和团队合作精神，能够接受 客户驻场工作；
7. 全职，北京地区，需具备在北京长期工作的条件；

优先考虑

• 计算机科学、信息技术或相关专业；
• 具备丰富的大规模分布式系统运维经验；
• 熟悉 Elasticsearch/Easysearch/OpenSearch 分片、路由、查询优化等高级功能；
• 拥有 Elastic Certified Engineer 认证；
• 具备大规模搜索引擎集群设计、扩展和调优经验；
• 熟悉其他搜索引擎技术（如 Solr、Lucene）者优先 ;
• 熟悉大数据处理相关技术（比如: Kafka 、Flink 等）者优先。

福利待遇

• 月薪: 15K - 25K，具体根据能力而定，试用期为 3-6 个月；
• 福利：五险一金/双休/国内法定假日/带薪年假等；

简历投递

简历投递：hello@infini.ltd（邮件标题请备注姓名+求职岗位+来自搜索客社区）
微信联系：INFINI-Labs（加微请备注求职岗位+来自搜索客社区）

欢迎有兴趣的小伙伴联系或推荐，期待您的加入！

今日快讯！就在刚刚，开源搜索领域行业巨头 Elastic 官方博客发表了一篇最新公告《Elasticsearch is Open Source, Again》，Elastic 创始人& CTO Shay Banon 宣布 Elasticsearch 和 Kibana 许可证协议修改为 AGPL。

以下为搜索客社区从 Elastic 官方博客翻译的原文内容：

Elasticsearch 再次成为开源软件

[D.N.A] Elasticsearch 和 Kibana 现在可以再次被称为开源软件了。这句话让我感到无比的兴奋。真的忍不住跳起来庆祝！我们所有在 Elastic 的人都很高兴。开源精神是我的DNA，也是 Elastic 的DNA。能够再次称 Elasticsearch 为开源软件，真的是一种纯粹的快乐。

[LOVE.] 简而言之，我们将在接下来的几周内，除了 ELv2 和 SSPL 之外，增加 AGPL 作为另一个许可选项。在更改许可后，我们从未停止过像一个开源社区一样相信和行动。但通过使用 AGPL 这一获得开放源码促进会（OSI）批准的许可，消除了人们可能存在的任何疑问或混淆。

[Not Like Us] 我们在 Elastic 从未停止过对开源的信仰。我个人对开源的信仰也从未动摇，至今已有 25 年了。那么为什么三年前我们要做出改变呢？我们遇到了与 AWS 相关的问题，以及他们的服务引发的市场混乱。在尝试了所有能想到的其他选项后，我们更改了许可协议，明知道这会导致 Elasticsearch 被分叉成另一个名称并走上不同的发展轨迹。这是一个漫长的故事。

[Like That] 好消息是，虽然过程痛苦，但结果奏效了。三年后，亚马逊已经完全投入到了他们的分叉项目中，市场的混乱（大部分）得到了缓解，我们与 AWS 的合作伙伴关系比以往更强。我们甚至被评为 AWS 的年度合作伙伴。我一直希望时间能过去得足够久，以至于我们可以安全地回到开源项目的状态——现在终于到了。

[All The Stars] 我们希望尽可能简化用户的使用体验。我们有用户非常喜欢 ELv2（一个受 BSD 启发的许可）。我们也有用户已经批准使用 SSPL（通过 MongoDB 使用）。这就是为什么我们只是增加了一个选项，而不是移除任何东西。如果你已经在使用并喜欢 Elasticsearch，请继续使用，没有任何变化。对于其他人，现在你也可以选择使用 AGPL。

[LOYALTY.] 我们选择 AGPL 而不是其他许可，是因为我们希望通过与 OSI 的合作，能在开源许可领域创造更多的选项。随着我们更改许可以来的发展（例如 Grafana 从 Apache2 转移到 AGPL），也许 AGPL 已经足够适用于像我们这样的基础设施软件了。我们致力于找到最合适的解决方案。

[euphoria] 我非常高兴能够再次称 Elasticsearch 为开源软件。

[Alright] 任何改变都可能引发混淆，当然也可能引来一些网络喷子。（网络喷子总是存在的，对吧？）让我们愉快地回答一些可能出现的问题吧。我可以想象到的一些问题如下，但我们会继续补充。

“更改许可是个错误，Elastic 现在在回撤”：我们在三年前更改许可时消除了很多市场混乱。由于我们的行动，很多事情已经改变。现在是一个完全不同的市场环境。我们不生活在过去。我们希望为用户打造一个更好的未来。正是因为我们当时采取了行动，现在我们才有能力采取新的行动。

“AGPL 不是真正的开源许可，X 才是”：AGPL 是获得 OSI 批准的许可，并且被广泛采用。例如，MongoDB 曾经使用 AGPL，Grafana 现在也是 AGPL。这表明 AGPL 并不影响使用或流行度。我们选择 AGPL 是因为我们认为这是与 OSI 一起为世界开辟更多开源道路的最佳方式，而不是减少开源。

“Elastic 更改许可是因为他们表现不好”：我要说的是，我今天对 Elastic 的未来依然充满期待。我为我们的产品和团队的执行力感到无比自豪。我们发布了无状态的 Elasticsearch ES|QL 和大量用于 GenAI 用例的向量数据库/混合搜索改进。我们在日志记录和可观察性方面大力投入 OTel。我们的安全产品 SIEM 不断添加令人惊叹的功能，并且是市场上增长最快的产品之一。用户的反应让我们感到非常谦卑。股市总有起伏，但我可以向你保证，我们始终着眼于长远发展，而这次的变更就是其中的一部分。

如果我们看到更多问题，会在上面继续添加，以期减少混淆。

[HUMBLE.] 为未来而构建真是令人兴奋。Elasticsearch 回归开源。万岁！这真是一件美妙的事情。今天真是美好的一天。

Forever :elasticheart: Open Source

Shay 2024-08-30

原文地址：https://www.elastic.co/blog/elasticsearch-is-open-source-again/

社区热评

Elasticsearch 再次回归开源的消息迅速引发了技术社区的广泛关注。这不仅是 Elastic 对自身开源信念的重申，也是其在激烈市场竞争中精心策划的一次战略调整。

三年前，Elastic 因与 AWS 的市场竞争而选择更改许可协议，这一决定在当时引发了不小的争议。尽管如此，事实证明，这一策略有效地减少了市场上的混淆，也为 Elastic 与 AWS 的合作奠定了更坚实的基础。如今，Elastic 再度选择开源，并新增 AGPL 作为许可选项，这一举措无疑展现了 Elastic 在市场中更加成熟的定位和对未来发展的自信。

这一变化不仅仅是一个公司的商业决策，更是开源生态系统的一次重要信号。Elastic 的回归开源，传递出一个明确的信息：即使在商业竞争中，开源仍然是企业实现长远发展的重要路径。随着这一决定的落地，其他软件公司可能也会重新审视自身的许可策略，推动更多开源项目的发展与创新。

此外，Elastic 选择 AGPL 作为新许可选项，也显示出其对开源生态未来走向的深刻洞察。AGPL 的引入，表明 Elastic 希望在开源社区中保持灵活性和多样性，同时推动整个行业向更加开放和透明的方向迈进。

总体而言，Elastic 重返开源的举动，不仅意在巩固其在开源社区中的地位，也为行业树立了一个新的标杆。这一事件无疑将成为开源软件发展史上的重要篇章，未来可能会激发更多企业重新考虑其开源战略，进而推动整个技术行业的进一步繁荣与进步。

让我们拭目以待！

故障现象

近日有客户找到我们，说有个 ES 集群节点，磁盘利用率达到了 82% ，而其节点才 63% ，想处理下这个节点，降低节点的磁盘利用率。

起初以为是没有打开自动平衡导致的，经查询，数据还是比较平衡的。 利用率较高的是 76 节点，如果 76 节点的分片比其他节点多，好像还比较合乎逻辑，但它反而比其他节点少了 12-15 个分片。那是 76 节点上的分片比较大？

索引情况

图中都是较大的索引，1 个索引 25TB 左右，共 160 个分片。

分片大小

节点 64 节点 77 节点 75 问题节点 76 可以看出分片大小没有出现较大的倾斜，分片大小和数据平衡的原因都被排除。

换个方向思考，节点 76 比其他节点多使用了磁盘空间 8 个 TB 左右，集群最大分片大小约 140GB ，8000/140=57 ，即节点 76 至少要比其他节点多 57 个分片才行，啊这...

会不会有其他的文件占用了磁盘空间？

我们登录到节点主机，排查是否有其他文件占用了磁盘空间。

结果：客户的数据路径是单独的数据磁盘，并没有其他文件，都是 ES 集群索引占用的空间。

现象总结

分片大小差不多的情况下，节点 76 的分片数还比别的节点还少 10 个左右，它的磁盘空间反而多占用了 8TB 。

这是不是太奇怪了？事出反常必有妖，继续往下查。

原因定位

通过进一步排查，我们发现节点 76 上有一批索引目录，在其他的节点上没有，而且也不在 GET \_cat/indices?v 命令的结果中。说明这些目录都是 dangling 索引占用的。

dangling 索引产生的原因

当 Elasticsearch 节点脱机时，如果删除的索引数量超过 Cluster.indes.tombstones.size，就会发生这种情况。

解决方案

通过命令删除 dangling 索引：

DELETE /\_dangling/<index-uuid>?accept_data_loss=true

最后

这次的分享就到这里了，欢迎与我一起交流 ES 的各种问题和解决方案。

关于极限科技（INFINI Labs）

极限科技，全称极限数据（北京）科技有限公司，是一家专注于实时搜索与数据分析的软件公司。旗下品牌极限实验室（INFINI Labs）致力于打造极致易用的数据探索与分析体验。

极限科技是一支年轻的团队，采用天然分布式的方式来进行远程协作，员工分布在全球各地，希望通过努力成为中国乃至全球企业大数据实时搜索分析产品的首选，为中国技术品牌输出添砖加瓦。

官网：https://infinilabs.cn

本次活动由 搜索客社区、极限科技（INFINI Labs）联合举办，活动主题将深入探讨 Elasticsearch 的两个核心方面：代码结构以及写入和查询的关键流程。本次活动将为 Elasticsearch 初学者和有经验的用户提供宝贵的见解，欢迎大家报名参加、交流学习。

活动主题：Elasticsearch 的代码结构和写入查询流程的解读
活动时间：2024 年 8 月 14 日 19:00-20:00（周三）
活动形式：微信视频号（极限实验室）直播
报名方式：关注或扫码海报中的二维码进行预约

嘉宾介绍

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

主题摘要

本次分享将探讨 Elasticsearch 的代码结构及其写入和查询流程。内容包括：项目架构、核心模块、插件系统，以及文档索引和查询的各个阶段与分布式查询协调。

活动亮点

1. 深入解析 Elasticsearch 代码结构：

   • 项目的整体结构：了解 Elasticsearch 项目的组织方式。
   • 核心模块及其职责：学习 Elasticsearch 的核心模块及其在系统中的具体角色。
   • 插件系统的设计：探索 Elasticsearch 灵活的插件系统设计及其扩展方式。

2. 详细解读 Elasticsearch 写入和查询流程：

   • 文档索引过程的各个阶段：跟随文档从初始接收至最终存储的索引过程。
   • 查询解析和执行的步骤：理解 Elasticsearch 中查询解析和执行的各个步骤。
   • 分布式查询的协调过程：学习 Elasticsearch 如何在分布式架构中协调查询，以提供高效且准确的搜索结果。

参与有奖

直播中将设有福袋抽奖环节，参与就有机会获得 INFINI Labs 周边纪念品，包括 T 恤、鸭舌帽、咖啡杯、指甲刀套件等等（图片仅供参考，款式、颜色与尺码随机）。

活动交流

活动交流群二维码 7 天内（8 月 12 日前）有效，如过期请添加小助手微信拉群。

本次 Meetup 是深入了解 Elasticsearch 内部工作机制、提升使用技能的绝佳机会。不要错过！

Meetup 讲师招募

搜索客社区 Meetup 的成功举办，离不开社区小伙伴的热情参与。目前社区讲师招募计划也在持续进行中，我们诚挚邀请各位技术大咖、行业精英踊跃提交演讲议题，与大家分享您的经验。

讲师报名链接：http://cfp.searchkit.cn
或扫描下方二维码，立刻报名成为讲师！

Meetup 活动聚焦 AI 与搜索领域的最新动态，以及数据实时搜索分析、向量检索、技术实践与案例分析、日志分析、安全等领域的深度探讨。

我们热切期待您的精彩分享！

往期回顾

【第 1 期】2024 搜索客 Meetup | Easysearch 结合大模型实现 RAG

关于 搜索客（SearchKit）社区

搜索客社区由 Elasticsearch 中文社区进行全新的品牌升级，以新的 Slogan：“搜索人自己的社区” 为宣言。汇集搜索领域最新动态、精选干货文章、精华讨论、文档资料、翻译与版本发布等，为广大搜索领域从业者提供更为丰富便捷的学习和交流平台。社区官网：https://searchkit.cn 。

在当今的数据驱动时代，搜索引擎的快照功能在数据保护和灾难恢复中至关重要。本文将对 Easysearch、Elasticsearch 和 Amazon OpenSearch 的快照兼容性进行比较，分析它们在快照创建、恢复、存储格式和跨平台兼容性等方面的特点，帮助大家更好地理解这些搜索引擎的差异，从而选择最适合自己需求的解决方案。

启动集群

Easysearch

服务器一般情况下默认参数都是很低的，而 Easysearch/Elasticsearch 是内存大户，所以就需要进行系统调优。

sysctl -w vm.max_map_count=262144

vm.max_map_count 是一个 Linux 内核参数，用于控制单个进程可以拥有的最大内存映射区域（VMA，Virtual Memory Areas）的数量。内存映射区域是指通过内存映射文件或匿名内存映射创建的虚拟内存区域。

这个参数在一些应用程序中非常重要，尤其是那些需要大量内存映射的应用程序，比如 Elasticsearch。Elasticsearch 使用内存映射文件来索引和搜索数据，这可能需要大量的内存映射区域。如果 vm.max_map_count 设置得太低，Elasticsearch 可能无法正常工作，并会出现错误信息。

调整 vm.max_map_count 参数的一些常见原因：

1. 支持大型数据集： 应用程序（如 Elasticsearch）在处理大型数据集时可能需要大量内存映射区域。增加 vm.max_map_count 可以确保这些应用程序有足够的内存映射区域来处理数据。

2. 防止内存错误： 如果 vm.max_map_count 设置得太低，当应用程序尝试创建超过限制的内存映射时，会出现错误，导致应用程序崩溃或无法正常工作。

3. 优化性能： 适当地设置 vm.max_map_count 可以优化应用程序的性能，确保内存映射操作顺利进行。

检查当前的 vm.max_map_count 值：

sysctl vm.max_map_count

或者查看 /proc/sys/vm/max_map_count 文件：

cat /proc/sys/vm/max_map_count

Elasticsearch 官方建议将 vm.max_map_count 设置为至少 262144。对于其他应用程序。

Easysearch 具体安装步骤见 INFINI Easysearch 尝鲜 Hands on

Amazon OpenSearch

使用 Amazon Web Services 控制台进行创建。

Elasticsearch

使用如下 docker compose 部署一个三节点的 ES 集群：

version: "2.2"
services:
  es01:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.10.2
    container_name: es01
    environment:
      - node.name=es01
      - cluster.name=es-docker-cluster
      - discovery.seed_hosts=es02,es03
      - cluster.initial_master_nodes=es01,es02,es03
      - bootstrap.memory_lock=true
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    ulimits:
      memlock:
        soft: -1
        hard: -1
    volumes:
      - data01:/usr/share/elasticsearch/data
    ports:
      - 9200:9200
    networks:
      - elastic
  es02:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.10.2
    container_name: es02
    environment:
      - node.name=es02
      - cluster.name=es-docker-cluster
      - discovery.seed_hosts=es01,es03
      - cluster.initial_master_nodes=es01,es02,es03
      - bootstrap.memory_lock=true
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    ulimits:
      memlock:
        soft: -1
        hard: -1
    volumes:
      - data02:/usr/share/elasticsearch/data
    networks:
      - elastic
  es03:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.10.2
    container_name: es03
    environment:
      - node.name=es03
      - cluster.name=es-docker-cluster
      - discovery.seed_hosts=es01,es02
      - cluster.initial_master_nodes=es01,es02,es03
      - bootstrap.memory_lock=true
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    ulimits:
      memlock:
        soft: -1
        hard: -1
    volumes:
      - data03:/usr/share/elasticsearch/data
    networks:
      - elastic

volumes:
  data01:
    driver: local
  data02:
    driver: local
  data03:
    driver: local

networks:
  elastic:
    driver: bridge

由于这个 docker compose 没有关于 kibana 的配置，所以我们还是用 Console 添加原生的 Elasticsearch 集群!

集群信息

快照还原的步骤

快照前的准备

插件安装

本次测试选择把索引快照备份到 Amazon S3，所以需要使用 S3 repository plugin，这个插件添加了对使用 Amazon S3 作为快照/恢复存储库的支持。

Easysearch 和 OpenSearch 集群自带了这个插件，所以无需额外安装。

对于自己部署的三节点 Elasticsearch 则需要进入每一个节点运行安装命令然后再重启集群，建议使用自动化运维工具来做这步，安装命令如下:

sudo bin/elasticsearch-plugin install repository-s3

如果不再需要这个插件，可以这样删除。

sudo bin/elasticsearch-plugin remove repository-s3

由于需要和 Amazon Web Services 打交道，所以我们需要设置 IAM 凭证，这个插件可以从 EC2 IAM instance profile，ECS task role 以及 EKS 的 Service account 读取相应的凭证。

对于托管的 Amazon OpenSearch 来说，我们无法在托管的 EC2 上绑定我们的凭证，所以需要新建一个 OpenSearchSnapshotRole，然后通过当前的用户把这个角色传递给服务，也就是我们说的 IAM:PassRole。

创建 OpenSearchSnapshotRole，策略如下：

{
  "Version": "2012-10-17",
  "Statement": [{
      "Action": [
        "s3:ListBucket"
      ],
      "Effect": "Allow",
      "Resource": [
        "arn:aws:s3:::bucket-name"
      ]
    },
    {
      "Action": [
        "s3:GetObject",
        "s3:PutObject",
        "s3:DeleteObject"
      ],
      "Effect": "Allow",
      "Resource": [
        "arn:aws:s3:::bucket-name/*"
      ]
    }
  ]
}

信任关系如下：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "es.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

然后在我们的 IAM user 上加上 PassRole 的权限，这样我们就可以把 OpenSearchSnapshotRole 传递给 OpenSearch 集群。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "iam:PassRole",
      "Resource": "arn:aws:iam::123456789012:role/OpenSearchSnapshotRole"
    }
  ]
}

注册存储库

在源集群执行注册

PUT /_snapshot/snapshot-repo-name
{
  "type": "s3",
  "settings": {
    "bucket": "<bucket-name>",
    "base_path": "<bucket-prefix>",

在目标集群同样执行这个语句，为了防止覆盖源集群存储库的数据，将 "readonly": true 添加到"settings" PUT 请求中，这样就只有一个集群具有对存储库的写入权限。

PUT /_snapshot/snapshot-repo-name
{
  "type": "s3",
  "settings": {
    "bucket": "<bucket-name>",
    "base_path": "<bucket-prefix>",
    "readonly": true,

对于 OpenSearch 来说，还需要执行 passrole，所以还需要添加role_arn这个字段，由于 IAM:PassRole 需要对 HTTP 请求做 signV4 日签名，所以这部常常使用 Postman 来完成。把角色传递过去之后，接下来的快照还原操作就可以在 OpenSearch Dashboard 中进行操作了。

需要注意的是，需要在 auth 这里输入 AccessKey，SecretKey，AWS Region，Service Name（es）来做 SignV4 的签名。

请求体如下：

{
  "type": "s3",
  "settings": {
    "bucket": "<bucket-name>",
    "base_path": "<bucket-prefix>",
    "readonly": true,
    "role_arn": "arn:aws:iam::123456789012:role/OpenSearchSnapshotRole"
  }
}

• 查看所有注册的存储库：
  • GET _snapshot：这个命令返回所有已注册的快照存储库列表及其基本信息。

GET _snapshot

{
  "es_repository": {
    "type": "s3",
    "settings": {
      "bucket": "your-s3-bucket-name",
      "region": "your-s3-bucket-region"
    }
  }
}

• 查看特定存储库的详细信息： GET _snapshot/es_repository：这个命令返回名为es_repository的存储库的详细配置信息，包括存储桶名称、区域和其他设置。

GET _snapshot/es_repository

{
  "es_repository": {
    "type": "s3",
    "settings": {
      "bucket": "your-s3-bucket-name",
      "region": "your-s3-bucket-region",
      "access_key": "your-access-key",
      "secret_key": "your-secret-key"
    }
  }
}

• 查看特定存储库中的快照： GET _cat/snapshots/es_repository?v：这个命令返回es_repository存储库中的所有快照及其详细信息，包括快照 ID、状态、开始时间、结束时间、持续时间、包含的索引数量、成功和失败的分片数量等。

GET _cat/snapshots/es_repository?v

id                     status start_epoch start_time end_epoch end_time duration indices successful_shards failed_shards total_shards
snapshot_1             SUCCESS 1628884800 08:00:00   1628888400 09:00:00 1h       3       10                0             10
snapshot_2             SUCCESS 1628971200 08:00:00   1628974800 09:00:00 1h       3       10                0             10

创建索引快照

# PUT _snapshot/my_repository/<my_snapshot_{now/d}>
PUT _snapshot/my_repository/my_snapshot
{
  "indices": "my-index,logs-my_app-default",
}

根据快照的大小不同，完成快照可能需要一些时间。默认情况下，create snapshot API 只会异步启动快照过程，该过程在后台运行。要更改为同步调用，可以将 wait_for_completion 查询参数设置为 true。

PUT _snapshot/my_repository/my_snapshot?wait_for_completion=true

另外还可以使用 clone snapshot API 克隆现有的快照。要监控当前正在运行的快照，可以使用带有 _current 请求路径参数的 get snapshot API。

GET _snapshot/my_repository/_current

如果要获取参与当前运行快照的每个分片的完整详细信息，可以使用 get snapshot status API。

GET _snapshot/_status

成功创建快照之后，就可以在 S3 上看到备份的数据块文件，这个是正确的快照层级结构：

需要注意的是， "base_path": ""这里最好不要加/，虽然不影响同集群迁移，这个会为我们在不同厂商的搜索引擎中迁移遇到问题，可能是这样的，所以需要注意。

所以在 Open Search 中还原 Elasticsearch 就遇到了这个问题：

{
  "error": {
    "root_cause": [
      {
        "type": "snapshot_missing_exception",
        "reason": "[easy_repository:2/-jOQ0oucQDGF3hJMNz-vKQ] is missing"
      }
    ],
    "type": "snapshot_missing_exception",
    "reason": "[easy_repository:2/-jOQ0oucQDGF3hJMNz-vKQ] is missing",
    "caused_by": {
      "type": "no_such_file_exception",
      "reason": "Blob object [11111/indices/7fv2zAi4Rt203JfsczUrBg/meta-YGnzxZABRBxW-2vqcmci.dat] not found: The specified key does not exist. (Service: S3, Status Code: 404, Request ID: R71DDHX4XXM0434T, Extended Request ID: d9M/HWvPvMFdPhB6KX+wYCW3ZFqeFo9EoscWPkulOXWa+TnovAE5PlemtuVzKXjlC+rrgskXAus=)"
    }
  },
  "status": 404
}

恢复索引快照

POST _snapshot/my_repository/my_snapshot_2099.05.06/_restore
{
  "indices": "my-index,logs-my_app-default",
}

各个集群的还原

1. Elasticsearch 7.10.2 的快照可以还原到 Easysearch 和 Amazon OpenSearch

2. 从 Easysearch 1.8.2 还原到 Elasticsearch 7.10.2 报错如下：

{
  "error": {
    "root_cause": [
      {
        "type": "snapshot_restore_exception",
        "reason": "[s3_repository:1/a2qV4NYIReqvgW6BX_nxxw] cannot restore index [my_indexs] because it cannot be upgraded"
      }
    ],
    "type": "snapshot_restore_exception",
    "reason": "[s3_repository:1/a2qV4NYIReqvgW6BX_nxxw] cannot restore index [my_indexs] because it cannot be upgraded",
    "caused_by": {
      "type": "illegal_state_exception",
      "reason": "The index [[my_indexs/ALlTCIr0RJqtP06ouQmf0g]] was created with version [1.8.2] but the minimum compatible version is [6.0.0-beta1]. It should be re-indexed in Elasticsearch 6.x before upgrading to 7.10.2."
    }
  },
  "status": 500
}

3. 从 Amazon OpenSearch 2.1.3 还原到 Elasticsearch 7.10.2 报错如下（无论是否开启兼容模式）：

{
  "error": {
    "root_cause": [
      {
        "type": "snapshot_restore_exception",
        "reason": "[aos:2/D-oyYSscSdCbZFcmPZa_yg] the snapshot was created with Elasticsearch version [36.34.78-beta2] which is higher than the version of this node [7.10.2]"
      }
    ],
    "type": "snapshot_restore_exception",
    "reason": "[aos:2/D-oyYSscSdCbZFcmPZa_yg] the snapshot was created with Elasticsearch version [36.34.78-beta2] which is higher than the version of this node [7.10.2]"
  },
  "status": 500
}

4. 从 Easysearch 1.8.2 还原到 Amazon OpenSearch2.13 报错如下（无论是否开启兼容模式）：

{
  "error": {
    "root_cause": [
      {
        "type": "snapshot_restore_exception",
        "reason": "[easy_repository:2/LE18AWHlRJu9rpz9BJatUQ] cannot restore index [my_indexs] because it cannot be upgraded"
      }
    ],
    "type": "snapshot_restore_exception",
    "reason": "[easy_repository:2/LE18AWHlRJu9rpz9BJatUQ] cannot restore index [my_indexs] because it cannot be upgraded",
    "caused_by": {
      "type": "illegal_state_exception",
      "reason": "The index [[my_indexs/VHOo7yfDTRa48uhQvquFzQ]] was created with version [1.8.2] but the minimum compatible version is OpenSearch 1.0.0 (or Elasticsearch 7.0.0). It should be re-indexed in OpenSearch 1.x (or Elasticsearch 7.x) before upgrading to 2.13.0."
    }
  },
  "status": 500
}

5. Amazon OpenSearch 还原到 Easysearch 同样失败

{
  "error": {
    "root_cause": [
      {
        "type": "snapshot_restore_exception",
        "reason": "[aoss:2/D-oyYSscSdCbZFcmPZa_yg] cannot restore index [aos] because it cannot be upgraded"
      }
    ],
    "type": "snapshot_restore_exception",
    "reason": "[aoss:2/D-oyYSscSdCbZFcmPZa_yg] cannot restore index [aos] because it cannot be upgraded",
    "caused_by": {
      "type": "illegal_state_exception",
      "reason": "The index [[aos/864WjTAXQCaxJ829V5ktaw]] was created with version [36.34.78-beta2] but the minimum compatible version is [6.0.0]. It should be re-indexed in Easysearch 6.x before upgrading to 1.8.2."
    }
  },
  "status": 500
}

6. Elasticsearch 8.14.3 迁移到 Amazon OpenSearch 或者 Elasticsearch 都是有这个报错：

{
  "error": {
    "root_cause": [
      {
        "type": "parsing_exception",
        "reason": "Failed to parse object: unknown field [uuid] found",
        "line": 1,
        "col": 25
      }
    ],
    "type": "repository_exception",
    "reason": "[snap] Unexpected exception when loading repository data",
    "caused_by": {
      "type": "parsing_exception",
      "reason": "Failed to parse object: unknown field [uuid] found",
      "line": 1,
      "col": 25
    }
  },
  "status": 500
}

这是由于 Elasticsearch 8 在创建快照的时候会默认加上一个 UUID 的字段，所以我们低版本的 Easysearch、Amazon OpenSearch 中会找不到这个字段，在执行GET _cat/snapshots/snap?v的时候就报错，及时在注册存储库的时候显示加上 UUID 的字段也无事无补。

{
  "snapshot-repo-name": {
    "type": "s3",
    "uuid": "qlJ0uqErRmW6aww2Fyt4Fg",
    "settings": {
    "bucket": "<bucket-name>",
    "base_path": "<bucket-prefix>",
    }
  },

以下是兼容性对比，每行第一列代表源集群，第一行代表目标集群：

Elasticsearch 的兼容列表官方的列表如下：

参考文献

1. 开始使用 Elastic Stack 和 Docker Compose：第 1 部分
   https://www.elastic.co/cn/blog/getting-started-with-the-elastic-stack-and-docker-compose

2. Docker Compose 部署多节点 Elasticsearch
   https://www.elastic.co/guide/en/elasticsearch/reference/7.10/docker.html#docker-compose-file

3. repository-s3 教程
   https://www.elastic.co/guide/en/elasticsearch/reference/8.14/repository-s3.html
   https://www.elastic.co/guide/en/elasticsearch/plugins/7.10/repository-s3.html

4. snapshot-restore
   https://www.elastic.co/guide/en/elasticsearch/reference/7.10/snapshot-restore.html

5. 在亚马逊 OpenSearch 服务中创建索引快照
   https://docs.amazonaws.cn/zh_cn/opensearch-service/latest/developerguide/managedomains-snapshots.html#managedomains-snapshot-restore

6. 教程：迁移至 Amazon OpenSearch Service
   https://docs.amazonaws.cn/zh_cn/opensearch-service/latest/developerguide/migration.html

关于 Easysearch 有奖征文活动

无论你是 Easysearch 的老用户，还是第一次听说这个名字，只要你对 INFINI Labs 旗下的 Easysearch 产品感兴趣，或者是希望了解 Easysearch，都可以参加这次活动。

详情查看：Easysearch 征文活动

作者：韩旭，亚马逊云技术支持，亚马逊云科技技领云博主，目前专注于云计算开发和大数据领域。

原文：https://infinilabs.cn/blog/2024/comparison-of-snapshot-compatibility-between-easysearch-elasticsearch-and-opensearch/

一、OpenAI 停服事件背景

OpenAI 自 2020 年推出 GPT-3 以来，在全球范围内引起了极大的反响。其强大的自然语言处理能力使其成为许多企业和开发者的首选工具。然而，2024 年 6 月 25 日，许多中国用户收到了一封来自 OpenAI 的邮件，邮件中明确表示，自 2024 年 7 月 9 日起，OpenAI 将停止对中国内地和香港地区提供 API 服务。

这一事件引发了国内开发者和企业的广泛讨论，特别是在人工智能技术应用逐渐深入的背景下，OpenAI 的停服无疑会对中国市场产生一定的影响。在 AI 技术迅猛发展的当下，许多中国企业和开发者依赖 OpenAI 的 API 进行各种应用的开发。尤其是在大模型技术领域，不少初创公司通过“套壳”OpenAI 技术快速推向市场。所谓“OpenAI 套壳”，是指一些公司仅对 OpenAI 的技术进行表面包装和小改动，而未进行深度创新 。OpenAI CEO 山姆·奥特曼曾明确指出，简单包装 OpenAI 技术的公司难以长久生存。此次 API 服务的终止，意味着这些企业需要寻找新的技术支持，或者在短时间内加速自主研发 。

二、国产替代的挑战与机遇

OpenAI 停止对中国提供 API 服务将对国内 AI 行业带来短期冲击，但从长远来看，这也可能成为推动国内 AI 技术自主创新和研发的契机。国内企业将被迫加大自主研发力度，减少对外部技术的依赖，从而推动国产 AI 技术的发展 。虽然目前国内大模型企业在技术上与 OpenAI 存在一定差距，但已有不少公司在不断赶超。例如，百度的文心大模型、科大讯飞的星火大模型以及清华智谱的 ChatGLM 基础模型，都在性能上逐渐逼近甚至超过了 GPT-4 。从现状来看，国内大模型厂商已经逐步进入了性能提升的关键期，应用场景从办公、生活拓展到医疗、工业、教育等领域。

三、Elasticsearch 国产化替代的需求

Elasticsearch（简称 ES）作为一种开源的分布式搜索和分析引擎，以其强大的搜索能力和高效的数据处理能力，广泛应用于各种大数据和搜索相关业务场景。然而，随着国际政治环境的变化和技术壁垒的加剧，依赖国外技术的风险日益凸显。在这种背景下，推动 Elasticsearch 的国产化替代至关重要，既能保障国家信息安全，也能促进国内技术自主创新和产业发展。

首先，依赖国外技术可能带来技术封锁风险，一旦国外企业因政治或经济原因停止服务，中国企业的业务连续性和数据安全性将受到威胁。推动国产化替代有助于规避这些风险，保障数据安全和业务的稳定运行。国外搜索引擎和数据库系统通常为西方市场优化，未必适合中国市场。例如，Elasticsearch 在处理中文信息时需要额外的插件和调整，而国产替代方案可以更好地适应本地化需求，同时确保数据存储和处理符合国家法律法规。

其次，推动 Elasticsearch 的国产化替代有助于激发国内企业的创新活力，减少对国外技术的依赖，提升在搜索和大数据领域的核心竞争力。采用国产替代方案不仅能够降低企业的技术成本，还能构建自主的技术生态系统，减少对外部的依赖。与此同时国产化替代也有助于提升国内技术人才水平，通过自主研发和技术创新培养出高水平的技术人才，推动技术创新和产业升级。

在性能和安全方面，Elasticsearch 在数据保护和性能方面存在不足。开源版本不具备数据保护功能，用户必须付费获得相关的安全功能。2021 年 Elastic 公司将其开源软件许可证变更为双授权许可，可能带来安全风险。此外，Elasticsearch 在数据读写性能和集群扩展上也存在技术挑战，进一步推动了国产替代的需求。

四、Easysearch：国产替代的优秀范例

在 OpenAI 退出中国市场的背景下，国产技术的重要性日益凸显。作为国产搜索引擎技术的代表，Easysearch 展示了强大的替代潜力。

Easysearch 是一款基于国内自主研发的高性能搜索引擎，其核心引擎基于开源的 Apache Lucene。与 Elasticsearch 相比，Easysearch 不仅提供了相当的功能，还进行了更具针对性的优化。

• 轻量级：Easysearch 的安装包小于 50MB，部署安装简单，适合企业快速上手。

• 跨平台支持：Easysearch 支持主流操作系统和 CPU 架构，兼容国产信创环境，确保多样化环境下的稳定运行。

• 高性能：Easysearch 针对不同的使用场景进行了优化，以更低的硬件成本提供更高的服务性能，从而实现降本增效。

• 安全增强：Easysearch 默认提供完整的企业级安全功能，支持 LDAP/AD 集成，能够对索引、文档和字段级别进行粒度化权限管控，确保数据的安全性和隐私性。

• 稳定可靠：Easysearch 修复了大量内核问题，解决了内存泄露、集群卡顿和查询缓慢等问题，经过严苛的业务环境考验，保证了其稳定性和可靠性。

• 简单易用：Easysearch 提供了企业级管理后台，运营标准化和自动化，使用户能够通过简单的页面操作实现专家级的运维管理工作。

Easysearch 特别注重数据安全和隐私保护，所有数据都在国内处理和存储，符合数据主权要求。这对关注数据安全和隐私的企业尤为重要，能够有效保障数据的安全性和合规性。

Easysearch 致力于构建一个开放的技术生态，通过丰富的开发者资源和支持，为国内开发者提供了一个强大的技术平台。

• 兼容性：Easysearch 兼容 Elasticsearch 的语法，支持 Elasticsearch 原有的 Query DSL 和 SQL 语法，并且兼容现有的 Elasticsearch SDK。这样，企业和开发者可以平滑地迁移到 Easysearch，无需修改现有代码。

• 中文适配：Easysearch 在功能上进行了中文适配，能够更好地支持中文搜索和分析，满足本地化需求。

• 功能增强：Easysearch 不断完善和增强企业级功能，使其在搜索业务场景中保持简洁与易用性，同时提升功能的丰富度和深度。

• 信创适配：Easysearch 支持信创环境，确保在国产环境中也能平稳运行和使用。

• 平滑迁移：Easysearch 基于网关实现了无缝的跨版本迁移与升级，用户可以随时安全回退，确保系统的持续稳定性。

Easysearch 作为国产搜索引擎技术的代表，展示了国产替代的强大潜力。通过自主研发和技术创新，Easysearch 不仅提供了强大的功能和性能，还确保了数据安全和隐私保护。它为国内企业和开发者提供了一个可靠的技术平台，推动了国产技术的发展和应用。未来，随着更多国产技术的崛起，中国将在全球技术领域占据更加重要的地位。

五、结束语

随着 OpenAI 服务在国内的终止，中国 AI 市场迎来了自主创新的机遇。国内企业如百度、科大讯飞等正加速自主研发，缩小技术差距。同时，在搜索引擎领域，极限科技推出 Easysearch 搜索引擎为 Elasticsearch 国产替代提供了新的选择。展望未来，国产技术的持续创新将助力中国在全球技术领域占据更重要的地位。

关于极限科技（INFINI Labs）

极限科技，全称极限数据（北京）科技有限公司，是一家专注于实时搜索与数据分析的软件公司。旗下品牌极限实验室（INFINI Labs）致力于打造极致易用的数据探索与分析体验。

极限科技是一支年轻的团队，采用天然分布式的方式来进行远程协作，员工分布在全球各地，希望通过努力成为中国乃至全球企业大数据实时搜索分析产品的首选，为中国技术品牌输出添砖加瓦。

官网：https://www.infinilabs.cn

作者：Muses

原文：https://infinilabs.cn/blog/2024/from-openai-service-discontinuation-to-the-chinese-market-opportunities-and-challenges-of-domestic-replacement-rising/