els学习

Elasticsearch 学习文档

目录

1. 简介
2. Elasticsearch 基础概念
   • 什么是 Elasticsearch
   • 核心特性
   • 应用场景
3. 安装与配置
   • 系统要求
   • 安装步骤
   • 基本配置
   • 启动与停止 Elasticsearch
4. Elasticsearch 架构
   • 节点（Node）
   • 集群（Cluster）
   • 索引（Index）
   • 分片（Shard）与副本（Replica）
5. 数据索引
   • 创建索引
   • 定义映射（Mapping）
   • 插入数据
   • 批量操作
6. 搜索与查询
   • 基本搜索
   • 过滤器（Filter）与查询（Query）
   • 布尔查询（Bool Query）
   • 匹配查询（Match Query）
   • 短语查询（Phrase Query）
7. 聚合（Aggregations）
   • 基本概念
   • 分桶聚合（Bucket Aggregations）
   • 指标聚合（Metric Aggregations）
   • 组合聚合
8. 高级功能
   • 自动完成（Autocomplete）
   • 相关性调整（Boosting）
   • 脚本查询（Script Query）
   • 地理位置搜索（Geo Search）
9. 管理与监控
   • 节点管理
   • 索引管理
   • 性能优化
   • 监控工具
10. 安全性
    • 用户认证与授权
    • 安全通信（SSL/TLS）
11. Elasticsearch API
    • RESTful API 概述
    • 常用 API 示例
12. 实践案例
    • 日志分析
    • 全文搜索应用
    • 实时数据分析
13. 资源与学习路径
    • 官方文档
    • 在线教程与课程
    • 社区与论坛
14. 附录: 常见问题与解决方案

1. 简介

什么是 Elasticsearch

Elasticsearch 是一个开源的、基于分布式的全文搜索和分析引擎。它由 Apache Lucene 构建而成，旨在提供高速、可扩展的搜索功能。Elasticsearch 以其强大的全文搜索能力和实时数据分析功能，被广泛应用于日志分析、监控系统、全文搜索应用等领域。

核心特性

• 分布式架构：支持水平扩展，可以处理大量数据和高并发请求。
• 实时搜索：几乎实时地将新数据纳入搜索范围。
• 全文搜索：基于倒排索引，提供高效的文本搜索能力。
• RESTful API：通过 HTTP/JSON 接口进行交互，易于集成。
• 聚合功能：支持复杂的数据聚合和分析。
• 多语言支持：内置多种语言分析器，支持多语言搜索。

应用场景

• 日志和监控：结合 Logstash 和 Kibana，构建 ELK 堆栈，用于日志收集、分析和可视化。
• 全文搜索：为网站、应用提供强大的搜索功能。
• 实时数据分析：实时处理和分析大量数据，生成业务洞察。
• 推荐系统：基于用户行为数据，进行实时推荐。

2. Elasticsearch 基础概念

核心组件

集群（Cluster）

一个 Elasticsearch 集群由一个或多个节点（Node）组成，集群中的所有节点共享相同的集群名称。集群负责管理数据分片和副本，并协调索引和搜索操作。

集群是一个由多个节点组成的整体，用于管理和存储数据，并处理所有的索引和搜索操作。可以把集群看作是一组节点的集合，它们通过合作来完成任务。

• 类比：
  • 如果把Elasticsearch比作一个图书馆，那么:
    • 集群就是整个图书馆系统，负责管理图书和读者服务。
    • 节点就是每个图书馆的分馆，它们各自存储和管理部分书籍，但整体上协同工作。
  • 例如，一个实际图书馆（集群）可能包含多个分馆（节点），读者可以在任何分馆借阅书籍。
• 集群特性：
  1. 唯一标识：每个集群都有一个唯一的名字(cluster.name)，用来区分不同的集群。
  2. 高可用性：即使一个节点离线，集群仍然可以继续工作（如果有足够的副本）。
  3. 扩展性：通过增加节点，可以水平扩展集群容量和性能。

节点（Node）

集群中的单个服务器实例称为节点。每个节点可以承担不同的角色，如主节点（Master Node）、数据节点（Data Node）、协调节点（Coordinator Node）等。

节点是Elasticsearch集群中的单个服务器实例。每个节点存储一部分数据，并参数处理搜索和索引请求。节点是集群的基本组成单位。

• 类比：
  • 在图书馆系统中：
    • 节点就是每一个分馆，负责存储部分书籍（数据）并服务读者。
    • 每个分馆可能专注于某些功能（比如，儿童读物馆、科技馆）。
  • 节点特性：
    1. 角色：节点可以有不同的角色，完成特定的任务。
    2. 通信：节点通过网络相互通信，协调分布式任务。
    3. 独立性：每个节点可以独立运行，但它们一起组成一个集群来提供红完成的服务。

索引（Index）

Elasticsearch 中的数据组织单位，相当于关系数据库中的数据库。每个索引包含一个或多个类型（Type），每种类型包含多个文档（Document）。从ES 7.X开始每个索引（Index）只能有一个类型（Type），让所有数据存储在单一逻辑分区中。这让Type的功能更接近一个表的逻辑结构。

应用：在应用中创建索引来组织和分类不同类型的数据。例如，一个电商应用可能有一个products索引用于存储产品信息，另一个customers索引用于存储客户信息。

索引是Elasticsearch中存储、组织和管理文档数据的逻辑单元，相当于关系数据库中的“表”。索引将文档的数据结构化存储，便于快速搜索和查询。

特点：

• 每个索引都有一个唯一的名字。
• 索引包含类型一致的文档（比如用户信息、订单数据等）。
• 索引通过分片（Shard）在多个节点上分布存储，提升性能和扩展性。

类比：

• 索引就像一个图书馆的书架，书架上的书都分类摆放，方便查找。
• 文档就是书架上的书，每本书都有自己的内容和编号。

创建索引示例：

PUT /my_index

或者带有自定义配置（比如分片和副本）

PUT /my_index
{
  "settings": {
    "number_of_shards": 3,
    "number_of_replicas": 1
  }
}

索引的用途：

• 存储结构化或半结构化的数据。
• 支持全文检索、高速查询和聚合分析。

文档（Document）

就像是关系型数据库中的一行数据，字段（Field）就像关系型数据库中的一列

索引中的基本数据单元，类似于关系数据库中的行。每个文档是一个 JSON 对象，包含一组字段（Field）。

应用：在应用中，会插入、更新、删除和查询文档、例如，向products索引中插入一个新的产品文档。

文档是Elasticsearch中数据的基本存储单元，存储实际信息的JSON格式对象。例如，每个用户、一条日志记录、一篇文章都可以是一个文档。

特点：

• 文档是无模式的：每个文档都可以包含不同的字段，Elasticsearch根据实际数据动态生成索引的映射（Mapping）。
• 每个文档通过一个唯一的ID标识，存储在某个索引中。

类比：

• 文档就像书架上的一本书，书中有章节（字段）和内容（值）。

文档示例：

{
  "id": 1,
  "title": "Elasticsearch Basics",
  "author": "John Doe",
  "published_date": "2024-01-01",
  "content": "Elasticsearch is a distributed search engine..."
}

操作文档：

1. 插入文档

   POST /my_index/_doc/1
   {
     "title": "Elasticsearch Basics",
     "author": "John Doe",
     "published_date": "2024-01-01",
     "content": "Elasticsearch is a distributed search engine..."
   }
   

2. 查询文档

   GET /my_index/_doc/1
   

3. 更新文档

   POST /my_index/_update/1
   {
     "doc": {
       "author": "Jane Doe"
     }
   }
   

字段（Field）

Field（字段）= 关系型数据库中的列

• Field 是 Elasticsearch 文档中的基本数据单元，类似于关系型数据库表中的一列。
• 每个文档（Document）由多个字段组成，字段的名称对应数据库表的列名，而字段的值对应列中的数据。

关系型数据库与 Elasticsearch 的对比

Field 的特点

1. 类型（Type）：
   每个 Field 都有一个类型（类似数据库中的列类型），常见类型包括：

   • text：适用于全文搜索的数据（类似字符串）。
   • keyword：适用于精确匹配的数据（类似唯一标识符）。
   • integer：整数类型。
   • float：浮点数类型。
   • date：日期类型。
   • boolean：布尔值类型。

2. 多字段（Multi-fields）：
   同一个 Field 可以存储多种类型的数据。例如：

   • 一个名为

     name
     

     的字段，可以作为全文搜索（

     text
     

     ）和精确匹配（

     keyword
     

     ）：

     "name": {
       "type": "text",
       "fields": {
         "keyword": {
           "type": "keyword"
         }
       }
     }
     

3. 动态映射和静态映射：

   • Elasticsearch 可以自动推断字段类型（动态映射）。
   • 也可以通过手动定义字段类型（静态映射）来提高查询效率和控制数据存储。

结合例子说明

假设有一个关系型数据库中的表如下：

在 Elasticsearch 中，可以表示为文档的 JSON 数据：

{
  "id": 1,
  "name": "Laptop",
  "price": 1200,
  "in_stock": true,
  "created_at": "2023-11-01T08:30:00Z"
}

这里的 Field 对应关系型数据库的 列：

• id：字段（整数类型）。
• name：字段（字符串类型，可以是text或keyword）。
• price：字段（浮点数类型）。
• in_stock：字段（布尔类型）。
• created_at：字段（日期类型）。

总结

在 Elasticsearch 中，Field 的作用相当于关系型数据库的 列，但它更灵活，可以适应不同的查询需求，比如：

• 提供全文搜索（text）。
• 精确匹配（keyword）。
• 结构化数据的范围查询（数字或日期类型）。

希望这个补充让你对 Field 的概念更加清晰！如果还有其他问题，可以随时问我。 😊

分片（Shard）与副本（Replica）

• 主分片（Primary Shard）：索引时数据的主要分片。
• 副本分片（Replica Shard）：主分片的副本，用于提高数据的高可用性和查询性能。
• 分片（Shard）：索引被划分为多个分片，以实现水平扩展。

分片是Elasticsearch中将索引拆分成更小单位的机制，用于分布式存储和处理数据。每个分片是一个独立的Lucense实例，可以存储索引数据的部分内容。

特点：

• 每个索引可以拆分成多个分片。
• 分片使得索引可以分布在多个节点上，从而支持大规模数据存储和高性能查询。

类比：

• 分片就像是图书馆中的一部分。
• 如果书架有10层，分片可以是每层上的书，它们一起组成完整的书架。

默认配置：

• 每个索引默认有1个分片和1个副本分片（可以在创建索引时指定）。

分片的作用;

1. 扩展性：
   • 分片允许将大索引分布到多个节点。
   • 增加节点时可以重分配分片，提高存储容量和性能。
2. 并行处理：
   • 查询是可以有多个分片同时处理，提高查询速度。

分片数量的配置：

分片数量在索引创建时设置，之后无法更改。通常分片数量由以下因素决定：

• 数据总量。
• 节点数量。
• 查询的并发需求。

副本（Replica)

什么是副本？

副本是分片的冗余拷贝，用于提升数据的高可用性和读取性能。

特点：

• 每个分片可以有多个副本分片。
• 副本分片存储在不同的节点上，防止单节点故障导致数据丢失。
• 副本分片不仅提供数据备份，还可以参与查询操作，分担读取压力。

类比：

• 副本就像图书馆的备份分馆。
• 如果某个分馆（主分片）无法使用，用户可以去备份分馆（副本分片）查找资料。

副本的作用：

1. 高可用性：
   • 如果主分片所在节点宕机，副本分片会接管任务，确保服务不会中断。
2. 读取优化：
   • 查询可以分发到多个副本分片，提升读取性能。

副本数量的配置：

副本数量在索引创建时配置，可以动态修改。例如，设置 1 个副本：

PUT /my_index/_settings
{
  "number_of_replicas": 1
}

分片与副本的关系

1. 分片是存储数据的基本单位，而副本是分片的备份。

2. 每个分片都有一个主分片和多个副本分片，主分片负责写入，副本分片用于备份和分担读取任务。

3. 数据存储示意图：

   • 一个索引有 3 个分片，每个分片有 1 个副本：

     • 主分片：shard_1, shard_2, shard_3
     • 副本分片：shard_1_replica, shard_2_replica, shard_3_replica

     节点1: shard_1, shard_2_replica
     节点2: shard_2, shard_3_replica
     节点3: shard_3, shard_1_replica
     

3. 安装与配置

系统要求

• 操作系统：Linux、macOS、Windows
• Java 版本：Elasticsearch 7.x 及以上内置了 OpenJDK，无需单独安装 Java。
• 内存：建议至少 4GB RAM，生产环境建议更高。
• CPU：多核处理器，提升并发处理能力。
• 磁盘：SSD 提升读写性能，保证充足的存储空间。

安装步骤

方法一：使用官方发行版

1. 下载 Elasticsearch

   前往 Elasticsearch 官方下载页面 下载适合您操作系统的版本。

2. 解压安装包

   tar -xzf elasticsearch-7.x.x-linux-x86_64.tar.gz
   cd elasticsearch-7.x.x/
   

3. 启动 Elasticsearch

   ./bin/elasticsearch
   

   默认情况下，Elasticsearch 会在 localhost:9200 监听 HTTP 请求。

方法二：使用 Docker

1. 拉取 Elasticsearch 镜像

   docker pull docker.elastic.co/elasticsearch/elasticsearch:7.x.x
   

2. 运行容器

   docker run -d --name elasticsearch -p 9200:9200 -p 9300:9300 -e "discovery.type=single-node" docker.elastic.co/elasticsearch/elasticsearch:7.x.x
   

基本配置

Elasticsearch 的配置文件位于 config/elasticsearch.yml。以下是一些常见的配置项：

# 集群名称
cluster.name: my-cluster

# 节点名称
node.name: node-1

# 数据存储路径
path.data: /var/lib/elasticsearch

# 日志路径
path.logs: /var/log/elasticsearch

# 网络配置
network.host: 0.0.0.0
http.port: 9200

# 分片与副本配置
index.number_of_shards: 3
index.number_of_replicas: 2

启动与停止 Elasticsearch

启动 Elasticsearch

# 在安装目录下
./bin/elasticsearch

停止 Elasticsearch

使用 Ctrl + C 终止进程，或通过系统服务管理工具停止。

4. Elasticsearch 架构

集群（Cluster）

一个集群由多个节点组成，所有节点共同协作完成数据存储和搜索任务。集群名称用于节点发现，确保同一集群中的节点能够互相识别和通信。

节点（Node）

主节点（Master Node）

负责集群管理任务，如索引创建、删除、分片分配等。主节点需要具备高可用性，通常配置多个主节点以避免单点故障。

数据节点（Data Node）

存储实际的索引数据，并执行数据相关操作，如 CRUD、搜索和聚合。数据节点的数量和配置直接影响集群的存储容量和查询性能。

协调节点（Coordinator Node）

协调和分发搜索请求，整合来自不同数据节点的结果，并返回给客户端。协调节点不存储数据，主要负责请求的分发和结果的聚合。

索引（Index）

索引是 Elasticsearch 中存储和管理数据的基本单位。每个索引包含多个文档，每个文档由多个字段组成。索引可以视为数据库中的表。

分片（Shard）与副本（Replica）

主分片（Primary Shard）

每个索引被划分为多个主分片，主分片负责存储索引数据。

副本分片（Replica Shard）

每个主分片可以有多个副本分片，用于提高数据的冗余性和查询性能。副本分片提供了数据备份，当主分片发生故障时，副本分片可以接管其工作。

分片策略

分片数量和副本数量在索引创建时指定。分片数量影响索引的并行处理能力和存储分布，副本数量影响数据的高可用性和查询性能。

5. 数据索引

创建索引

使用 Elasticsearch 的 RESTful API 创建索引。可以通过 PUT 请求指定索引名称和配置。

示例：创建一个名为 products 的索引

curl -X PUT "localhost:9200/products" -H 'Content-Type: application/json' -d'
{
  "settings": {
    "number_of_shards": 3,
    "number_of_replicas": 2
  },
  "mappings": {
    "properties": {
      "product_id": { "type": "long" },
      "name": { "type": "text" },
      "description": { "type": "text" },
      "price": { "type": "double" },
      "available": { "type": "boolean" },
      "tags": { "type": "keyword" },
      "created_at": { "type": "date" }
    }
  }
}
'

定义映射（Mapping）

映射定义了索引中文档字段的类型和属性。正确的映射有助于提升搜索和分析性能。

常见字段类型

• Text：用于全文搜索，支持分词。
• Keyword：用于精确匹配，不分词。
• Long/Integer/Double/Float：数值类型。
• Boolean：布尔类型。
• Date：日期类型。
• Geo-point：地理位置类型。
• Nested：嵌套类型，用于复杂对象。

示例：定义 products 索引的映射

{
  "mappings": {
    "properties": {
      "product_id": { "type": "long" },
      "name": { "type": "text" },
      "description": { "type": "text" },
      "price": { "type": "double" },
      "available": { "type": "boolean" },
      "tags": { "type": "keyword" },
      "created_at": { "type": "date" }
    }
  }
}

插入数据

使用 POST 或 PUT 请求向索引中添加文档。

示例：向 products 索引插入一个文档

curl -X POST "localhost:9200/products/_doc/1" -H 'Content-Type: application/json' -d'
{
  "product_id": 1,
  "name": "Wireless Mouse",
  "description": "A comfortable wireless mouse with ergonomic design.",
  "price": 25.99,
  "available": true,
  "tags": ["electronics", "computer accessories"],
  "created_at": "2023-01-15"
}
'

批量操作

使用 _bulk API 批量插入、更新或删除文档，提高操作效率。

示例：批量插入多个文档

curl -X POST "localhost:9200/products/_bulk" -H 'Content-Type: application/json' -d'
{ "index": { "_id": 2 } }
{ "product_id": 2, "name": "Gaming Keyboard", "description": "Mechanical keyboard with RGB lighting.", "price": 79.99, "available": true, "tags": ["electronics", "computer accessories"], "created_at": "2023-02-20" }
{ "index": { "_id": 3 } }
{ "product_id": 3, "name": "USB-C Hub", "description": "Multiport adapter with HDMI and USB 3.0 ports.", "price": 45.50, "available": false, "tags": ["electronics", "computer accessories"], "created_at": "2023-03-05" }
'

6. 搜索与查询

基本搜索

使用 GET 请求进行简单的全文搜索。

示例：在 products 索引中搜索包含 “wireless” 的文档

curl -X GET "localhost:9200/products/_search" -H 'Content-Type: application/json' -d'
{
  "query": {
    "match": {
      "name": "wireless"
    }
  }
}
'

过滤器（Filter）与查询（Query）

• 查询（Query）：用于评分和相关性排序，适用于全文搜索。
• 过滤器（Filter）：用于精确匹配和布尔条件，不影响评分，适用于结构化数据筛选。

布尔查询（Bool Query）

结合多个查询条件，实现复杂的搜索逻辑。

示例：搜索价格在 20 到 50 之间，且库存可用的产品

curl -X GET "localhost:9200/products/_search" -H 'Content-Type: application/json' -d'
{
  "query": {
    "bool": {
      "must": {
        "match": { "name": "mouse" }
      },
      "filter": [
        { "range": { "price": { "gte": 20, "lte": 50 } } },
        { "term": { "available": true } }
      ]
    }
  }
}
'

匹配查询（Match Query）

用于全文搜索，基于分析器进行分词匹配。

示例：匹配描述中包含 “ergonomic design” 的产品

curl -X GET "localhost:9200/products/_search" -H 'Content-Type: application/json' -d'
{
  "query": {
    "match": {
      "description": "ergonomic design"
    }
  }
}
'

短语查询（Phrase Query）

精确匹配连续的词语，适用于短语搜索。

示例：匹配名称中包含 “wireless mouse” 的产品

curl -X GET "localhost:9200/products/_search" -H 'Content-Type: application/json' -d'
{
  "query": {
    "match_phrase": {
      "name": "wireless mouse"
    }
  }
}
'

7. 聚合（Aggregations）

基本概念

聚合是 Elasticsearch 中用于数据分析的强大功能，类似于 SQL 中的 GROUP BY 和聚合函数。通过聚合，可以计算文档的统计信息、分组数据等。

分桶聚合（Bucket Aggregations）

将文档分组到不同的桶中，每个桶代表一个分组条件。

示例：按标签（tags）分组，统计每个标签的文档数量

curl -X GET "localhost:9200/products/_search" -H 'Content-Type: application/json' -d'
{
  "size": 0,
  "aggs": {
    "tags_count": {
      "terms": { "field": "tags.keyword", "size": 10 }
    }
  }
}
'

指标聚合（Metric Aggregations）

对文档中的数值字段进行统计计算，如求和、平均值等。

示例：计算所有产品的平均价格

curl -X GET "localhost:9200/products/_search" -H 'Content-Type: application/json' -d'
{
  "size": 0,
  "aggs": {
    "average_price": {
      "avg": { "field": "price" }
    }
  }
}
'

组合聚合

结合多种聚合类型，实现复杂的数据分析。

示例：按标签分组，统计每个标签下的平均价格

curl -X GET "localhost:9200/products/_search" -H 'Content-Type: application/json' -d'
{
  "size": 0,
  "aggs": {
    "tags_group": {
      "terms": { "field": "tags.keyword", "size": 10 },
      "aggs": {
        "average_price": { "avg": { "field": "price" } }
      }
    }
  }
}
'

8. 高级功能

自动完成（Autocomplete）

实现输入建议功能，提升用户搜索体验。

示例：使用 completion Suggesters

1. 定义映射

   在索引创建时，定义 completion 类型的字段。

   {
     "mappings": {
       "properties": {
         "name": { "type": "text" },
         "suggest": { "type": "completion" }
       }
     }
   }
   

2. 插入数据

   curl -X POST "localhost:9200/products/_doc/1" -H 'Content-Type: application/json' -d'
   {
     "name": "Wireless Mouse",
     "suggest": {
       "input": ["Wireless Mouse", "Mouse", "Wireless"]
     }
   }
   '
   

3. 查询建议

   curl -X GET "localhost:9200/products/_search" -H 'Content-Type: application/json' -d'
   {
     "suggest": {
       "product-suggest": {
         "prefix": "wir",
         "completion": {
           "field": "suggest",
           "fuzzy": {
             "fuzziness": 2
           }
         }
       }
     }
   }
   '
   

相关性调整（Boosting）

提升特定字段或条件的相关性，影响搜索结果的排序。

示例：提升名称（name）字段的匹配权重

curl -X GET "localhost:9200/products/_search" -H 'Content-Type: application/json' -d'
{
  "query": {
    "bool": {
      "should": [
        { "match": { "name": { "query": "mouse", "boost": 2 } } },
        { "match": { "description": "mouse" } }
      ]
    }
  }
}
'

脚本查询（Script Query）

使用脚本语言（如 Painless）进行复杂的查询和计算。

示例：查找价格大于平均价格的产品

1. 计算平均价格

   curl -X GET "localhost:9200/products/_search" -H 'Content-Type: application/json' -d'
   {
     "size": 0,
     "aggs": {
       "average_price": { "avg": { "field": "price" } }
     }
   }
   '
   

2. 使用脚本查询

   假设平均价格为 50.0。

   curl -X GET "localhost:9200/products/_search" -H 'Content-Type: application/json' -d'
   {
     "query": {
       "script": {
         "script": {
           "source": "doc['price'].value > params.avg_price",
           "params": { "avg_price": 50.0 }
         }
       }
     }
   }
   '
   

地理位置搜索（Geo Search）

基于地理位置的数据搜索，适用于位置相关的应用场景。

示例：查找距离某点 10 公里以内的产品

1. 定义映射

   {
     "mappings": {
       "properties": {
         "location": { "type": "geo_point" }
       }
     }
   }
   

2. 插入数据

   curl -X POST "localhost:9200/products/_doc/1" -H 'Content-Type: application/json' -d'
   {
     "name": "Wireless Mouse",
     "location": { "lat": 40.7128, "lon": -74.0060 }
   }
   '
   

3. 执行地理位置搜索

   curl -X GET "localhost:9200/products/_search" -H 'Content-Type: application/json' -d'
   {
     "query": {
       "bool": {
         "filter": {
           "geo_distance": {
             "distance": "10km",
             "location": { "lat": 40.7128, "lon": -74.0060 }
           }
         }
       }
     }
   }
   '
   

9. 管理与监控

节点管理

查看集群健康状态

curl -X GET "localhost:9200/_cluster/health?pretty"

查看集群信息

curl -X GET "localhost:9200/_cluster/state?pretty"

索引管理

查看所有索引

curl -X GET "localhost:9200/_cat/indices?v"

删除索引

curl -X DELETE "localhost:9200/products"

性能优化

调整分片与副本

根据数据量和查询需求，调整索引的分片数量和副本数量。

使用缓存

Elasticsearch 内置了查询缓存和过滤缓存，通过优化缓存配置，可以提升查询性能。

压缩与归档

定期对不常用的索引进行压缩或归档，节省存储空间，提高集群性能。

监控工具

• Kibana：Elasticsearch 的可视化工具，用于监控集群状态、查看索引、执行搜索等。
• Elasticsearch Monitoring：通过 X-Pack 提供的监控功能，实时监控集群的性能指标。
• 第三方工具：如 Grafana 配合 Prometheus，用于构建自定义的监控面板。

10. 安全性

用户认证与授权

通过 X-Pack 提供的安全功能，配置用户认证和角色授权。

示例：创建用户和分配角色

1. 启用安全功能

   在 elasticsearch.yml 中启用安全功能。

   xpack.security.enabled: true
   

2. 设置密码

   ./bin/elasticsearch-setup-passwords interactive
   

3. 创建角色

   curl -X POST "localhost:9200/_security/role/my_role" -H 'Content-Type: application/json' -d'
   {
     "cluster": ["all"],
     "indices": [
       {
         "names": [ "products" ],
         "privileges": ["read", "write"]
       }
     ]
   }
   '
   

4. 创建用户并分配角色

   curl -X POST "localhost:9200/_security/user/john" -H 'Content-Type: application/json' -d'
   {
     "password" : "john_password",
     "roles" : [ "my_role" ],
     "full_name" : "John Doe",
     "email" : "john.doe@example.com"
   }
   '
   

安全通信（SSL/TLS）

通过配置 SSL/TLS，加密 Elasticsearch 集群内的通信，保护数据传输的安全性。

示例：生成自签名证书

使用 Elasticsearch Certificate Tool 生成证书。

./bin/elasticsearch-certutil cert -out config/certs/elastic-certificates.p12 -pass ""

配置 SSL/TLS

在 elasticsearch.yml 中配置 SSL/TLS。

xpack.security.transport.ssl.enabled: true
xpack.security.transport.ssl.verification_mode: certificate
xpack.security.transport.ssl.keystore.path: certs/elastic-certificates.p12
xpack.security.transport.ssl.truststore.path: certs/elastic-certificates.p12

xpack.security.http.ssl.enabled: true
xpack.security.http.ssl.keystore.path: certs/elastic-certificates.p12

11. Elasticsearch API

RESTful API 概述

Elasticsearch 提供了丰富的 RESTful API，允许通过 HTTP 方法（如 GET、POST、PUT、DELETE）进行索引、搜索、管理等操作。

常用 API 示例

索引操作

• 创建索引

  curl -X PUT "localhost:9200/products" -H 'Content-Type: application/json' -d'
  {
    "settings": { ... },
    "mappings": { ... }
  }
  '
  

• 插入文档

  curl -X POST "localhost:9200/products/_doc/1" -H 'Content-Type: application/json' -d'
  {
    "product_id": 1,
    "name": "Wireless Mouse",
    ...
  }
  '
  

• 获取文档

  curl -X GET "localhost:9200/products/_doc/1"
  

• 更新文档

  curl -X POST "localhost:9200/products/_doc/1/_update" -H 'Content-Type: application/json' -d'
  {
    "doc": { "price": 29.99 }
  }
  '
  

• 删除文档

  curl -X DELETE "localhost:9200/products/_doc/1"
  

搜索操作

• 基本搜索

  curl -X GET "localhost:9200/products/_search" -H 'Content-Type: application/json' -d'
  {
    "query": { "match": { "name": "mouse" } }
  }
  '
  

• 聚合查询

  curl -X GET "localhost:9200/products/_search" -H 'Content-Type: application/json' -d'
  {
    "size": 0,
    "aggs": { "average_price": { "avg": { "field": "price" } } }
  }
  '
  

集群管理 API

• 查看集群健康状态

  curl -X GET "localhost:9200/_cluster/health?pretty"
  

• 查看集群状态

  curl -X GET "localhost:9200/_cluster/state?pretty"
  

• 查看集群节点信息

  curl -X GET "localhost:9200/_cat/nodes?v"
  

12. 实践案例

日志分析

结合 Logstash 和 Kibana，构建 ELK 堆栈，实现日志的收集、存储、分析和可视化。

步骤

1. 安装 Logstash： 安装 Logstash

2. 配置 Logstash： 定义输入、过滤器和输出。

   input {
     file {
       path => "/var/log/myapp/*.log"
       start_position => "beginning"
     }
   }
   
   filter {
     grok {
       match => { "message" => "%{COMMONAPACHELOG}" }
     }
     date {
       match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
     }
   }
   
   output {
     elasticsearch {
       hosts => ["localhost:9200"]
       index => "myapp-logs-%{+YYYY.MM.dd}"
     }
     stdout { codec => rubydebug }
   }
   

3. 启动 Logstash：

   bin/logstash -f logstash.conf
   

4. 使用 Kibana 可视化日志：

   在 Kibana 中配置索引模式，创建仪表板进行数据可视化。

全文搜索应用

为电商网站实现产品搜索功能，支持关键词匹配和语义搜索。

步骤

1. 创建产品索引

   curl -X PUT "localhost:9200/products" -H 'Content-Type: application/json' -d'
   {
     "settings": { "number_of_shards": 3, "number_of_replicas": 1 },
     "mappings": {
       "properties": {
         "product_id": { "type": "long" },
         "name": { "type": "text" },
         "description": { "type": "text" },
         "price": { "type": "double" },
         "available": { "type": "boolean" },
         "tags": { "type": "keyword" },
         "created_at": { "type": "date" }
       }
     }
   }
   '
   

2. 插入产品数据

   curl -X POST "localhost:9200/products/_doc/1" -H 'Content-Type: application/json' -d'
   {
     "product_id": 1,
     "name": "Wireless Mouse",
     "description": "A comfortable wireless mouse with ergonomic design.",
     "price": 25.99,
     "available": true,
     "tags": ["electronics", "computer accessories"],
     "created_at": "2023-01-15"
   }
   '
   

3. 实现搜索功能

   在前端应用中，通过调用 Elasticsearch 的搜索 API，实现搜索功能。结合 JavaScript 框架（如 React、Vue）和后端服务（如 Node.js、Python），构建交互式搜索界面。

15. 使用 Python 操作 Elasticsearch

在本节中，我们将通过 Python 实现对 Elasticsearch 的实际操作，包括连接、创建索引、插入数据、查询数据等。我们将使用官方提供的 elasticsearch-py 客户端，这是与 Elasticsearch 交互的推荐方式。

15.1 安装 Elasticsearch Python 客户端

首先，确保你已经安装了 Python（建议使用 Python 3.6 及以上版本）。然后使用 pip 安装 elasticsearch 客户端库。

pip install elasticsearch

15.2 连接到 Elasticsearch

使用 elasticsearch-py 客户端连接到 Elasticsearch 集群。默认情况下，Elasticsearch 运行在 localhost:9200。

from elasticsearch import Elasticsearch

# 创建一个 Elasticsearch 客户端实例，连接到本地的 Elasticsearch 服务
es = Elasticsearch([{'host': 'localhost', 'port': 9200}])

# 使用 ping 方法检查是否能够成功连接到 Elasticsearch
if es.ping():
    print("连接成功")
else:
    print("连接失败")

输出示例：

连接成功

详细说明：

• 导入 Elasticsearch 类：首先从 elasticsearch 库中导入 Elasticsearch 类，这是与 Elasticsearch 交互的主要接口。
• 创建客户端实例：通过提供主机名和端口号，创建一个 Elasticsearch 客户端实例。这里假设 Elasticsearch 运行在本地主机（localhost）的默认端口（9200）。
• 检查连接：使用 ping() 方法发送一个简单的请求到 Elasticsearch 以验证连接是否成功。如果连接成功，输出 “连接成功”；否则，输出 “连接失败”。

15.3 创建索引

使用 Python 创建一个新的索引。可以在创建索引时指定索引的设置和映射。

# 定义索引名称
index_name = 'products'

# 定义索引的设置和映射
index_body = {
    "settings": {
        "number_of_shards": 3,    # 设置主分片数量为3
        "number_of_replicas": 1   # 设置副本分片数量为1
    },
    "mappings": {
        "properties": {
            "product_id": { "type": "long" },           # 定义 product_id 为长整数类型
            "name": { "type": "text" },                # 定义 name 为文本类型，适用于全文搜索
            "description": { "type": "text" },         # 定义 description 为文本类型
            "price": { "type": "double" },             # 定义 price 为双精度浮点数类型
            "available": { "type": "boolean" },        # 定义 available 为布尔类型
            "tags": { "type": "keyword" },             # 定义 tags 为关键字类型，适用于精确匹配
            "created_at": { "type": "date" }           # 定义 created_at 为日期类型
        }
    }
}

# 检查索引是否已存在
if not es.indices.exists(index=index_name):
    # 如果索引不存在，则创建索引
    es.indices.create(index=index_name, body=index_body)
    print(f"索引 '{index_name}' 创建成功")
else:
    print(f"索引 '{index_name}' 已存在")

输出示例：

索引 'products' 创建成功

详细说明：

• 定义索引名称：这里我们创建一个名为 products 的索引，用于存储产品信息。
• 定义索引设置和映射：
  • settings：
    • number_of_shards：设置主分片的数量。分片是 Elasticsearch 分布式存储的基础单位，主分片数量决定了索引的数据将被分割成多少部分。
    • number_of_replicas：设置每个主分片的副本数量。副本用于提高数据的冗余性和查询性能。
  • mappings：
    • 定义每个字段的类型，以便 Elasticsearch 正确地存储和索引数据。
• 检查索引是否存在：使用 es.indices.exists 方法检查索引是否已经存在，避免重复创建。
• 创建索引：如果索引不存在，使用 es.indices.create 方法创建索引，并传入前面定义的 index_body。
• 输出结果：根据索引是否已存在，输出相应的消息。

15.4 插入文档

向创建的索引中插入单个文档。

# 定义文档数据
doc = {
    "product_id": 1,
    "name": "Wireless Mouse",
    "description": "A comfortable wireless mouse with ergonomic design.",
    "price": 25.99,
    "available": True,
    "tags": ["electronics", "computer accessories"],
    "created_at": "2023-01-15"
}

# 使用 index 方法将文档插入到指定索引中，指定文档的唯一 ID 为1
res = es.index(index=index_name, id=1, document=doc)

# 输出操作结果，查看文档是否成功创建
print(res['result'])  # 可能的输出包括 'created' 或 'updated'

输出示例：

created

详细说明：

• 定义文档数据：创建一个包含产品信息的字典 doc，包括产品ID、名称、描述、价格、可用性、标签和创建日期。
• 插入文档：
  • 使用 es.index 方法将文档插入到 products 索引中。
  • 指定文档的唯一标识符 id 为1。如果ID已存在，则该操作将更新现有文档。
• 输出结果：通过打印 res['result']，可以看到操作结果，通常为 'created' 表示新文档已创建，或 'updated' 表示现有文档已更新。

15.5 批量插入文档

使用 _bulk API 批量插入多个文档，提高插入效率。

from elasticsearch import helpers

# 定义多个文档，准备批量插入
docs = [
    {
        "_index": index_name,        # 指定目标索引
        "_id": 2,                    # 指定文档ID为2
        "_source": {                 # 文档内容
            "product_id": 2,
            "name": "Gaming Keyboard",
            "description": "Mechanical keyboard with RGB lighting.",
            "price": 79.99,
            "available": True,
            "tags": ["electronics", "computer accessories"],
            "created_at": "2023-02-20"
        }
    },
    {
        "_index": index_name,
        "_id": 3,
        "_source": {
            "product_id": 3,
            "name": "USB-C Hub",
            "description": "Multiport adapter with HDMI and USB 3.0 ports.",
            "price": 45.50,
            "available": False,
            "tags": ["electronics", "computer accessories"],
            "created_at": "2023-03-05"
        }
    },
    # 可以继续添加更多文档
]

# 使用 helpers.bulk 方法批量插入文档
helpers.bulk(es, docs)
print("批量插入完成")

输出示例：

批量插入完成

详细说明：

• 导入 helpers 模块：从 elasticsearch 库中导入 helpers 模块，该模块提供了批量操作的辅助函数。
• 定义多个文档：创建一个包含多个文档的列表 docs，每个文档包括 _index（目标索引名）、_id（文档ID）和 _source（文档内容）。
• 批量插入文档：
  • 使用 helpers.bulk 方法将 docs 列表中的所有文档一次性插入到 Elasticsearch 中。
  • 这种方式比逐个插入更高效，特别是当需要插入大量数据时。
• 输出结果：插入完成后，打印 “批量插入完成” 以确认操作结束。

15.6 查询文档

使用 search API 进行查询操作，例如基本匹配查询和布尔查询。

15.6.1 基本匹配查询

搜索名称中包含 “wireless” 的产品。

# 定义查询语句，使用 match 查询在 'name' 字段中查找包含 'wireless' 的文档
query = {
    "query": {
        "match": {
            "name": "wireless"
        }
    }
}

# 执行搜索操作，指定索引和查询体
res = es.search(index=index_name, body=query)

# 处理搜索结果
print(f"找到 {res['hits']['total']['value']} 个匹配的文档")
for hit in res['hits']['hits']:
    # 打印每个匹配文档的内容
    print(hit["_source"])

输出示例：

找到 1 个匹配的文档
{'product_id': 1, 'name': 'Wireless Mouse', 'description': 'A comfortable wireless mouse with ergonomic design.', 'price': 25.99, 'available': True, 'tags': ['electronics', 'computer accessories'], 'created_at': '2023-01-15'}

详细说明：

• 定义查询语句：使用 match 查询在 name 字段中搜索包含关键词 “wireless” 的文档。match 查询适用于全文搜索，基于分析器进行分词匹配。
• 执行搜索：调用 es.search 方法，传入索引名称和查询语句，执行搜索操作。
• 处理结果：
  • res['hits']['total']['value']：获取匹配文档的总数量。
  • res['hits']['hits']：获取实际匹配的文档列表。
  • 循环遍历每个匹配的文档，并打印其内容。

15.6.2 布尔查询

搜索价格在 20 到 50 之间，且库存可用的产品。

# 定义布尔查询，结合 must 和 filter 条件
query = {
    "query": {
        "bool": {
            "must": {
                "match": { "name": "mouse" }    # 必须匹配 'name' 字段包含 'mouse'
            },
            "filter": [
                { "range": { "price": { "gte": 20, "lte": 50 } } },   # 价格在20到50之间
                { "term": { "available": True } }                    # 'available' 字段为 True
            ]
        }
    }
}

# 执行搜索操作
res = es.search(index=index_name, body=query)

# 处理搜索结果
print(f"找到 {res['hits']['total']['value']} 个匹配的文档")
for hit in res['hits']['hits']:
    print(hit["_source"])

输出示例：

找到 1 个匹配的文档
{'product_id': 1, 'name': 'Wireless Mouse', 'description': 'A comfortable wireless mouse with ergonomic design.', 'price': 25.99, 'available': True, 'tags': ['electronics', 'computer accessories'], 'created_at': '2023-01-15'}

详细说明：

• 定义布尔查询：
  • must：必须匹配的查询条件，这里要求 name 字段中包含 “mouse”。
  • **filter：**过滤条件，不影响评分，仅用于筛选结果。这包括：
    • range 查询：筛选 price 字段在 20 到 50 之间的文档。
    • term 查询：筛选 available 字段为 True 的文档。
• 执行搜索：调用 es.search 方法，传入索引名称和布尔查询，执行搜索操作。
• 处理结果：与基本匹配查询类似，打印匹配文档的数量和内容。

15.7 聚合查询

使用聚合功能进行数据分析，例如统计平均价格和按标签分组。

15.7.1 计算平均价格

# 定义聚合查询，计算所有文档的平均价格
query = {
    "size": 0,  # 设置为0，不返回具体文档，只返回聚合结果
    "aggs": {
        "average_price": {
            "avg": { "field": "price" }    # 使用 avg 聚合计算 'price' 字段的平均值
        }
    }
}

# 执行聚合查询
res = es.search(index=index_name, body=query)

# 处理聚合结果，获取平均价格的值
avg_price = res['aggregations']['average_price']['value']
print(f"平均价格: {avg_price}")

输出示例：

平均价格: 50.83

详细说明：

• 定义聚合查询：
  • size：设置为0，表示不返回实际的文档数据，只返回聚合结果，节省带宽和处理时间。
  • aggs：定义聚合操作，这里使用 avg 聚合计算 price 字段的平均值，并命名为 average_price。
• 执行聚合查询：调用 es.search 方法，传入索引名称和聚合查询，执行聚合操作。
• 处理结果：从响应中提取 average_price 的值，并打印出来。

15.7.2 按标签分组统计

统计每个标签的文档数量。

# 定义聚合查询，按 'tags.keyword' 字段进行分组，并统计每组的文档数量
query = {
    "size": 0,  # 不返回具体文档，只返回聚合结果
    "aggs": {
        "tags_count": {
            "terms": { 
                "field": "tags.keyword",  # 使用 'tags.keyword' 字段进行分组
                "size": 10                # 返回前10个标签
            }
        }
    }
}

# 执行聚合查询
res = es.search(index=index_name, body=query)

# 处理聚合结果，遍历每个标签及其对应的文档数量
for bucket in res['aggregations']['tags_count']['buckets']:
    print(f"标签: {bucket['key']}, 数量: {bucket['doc_count']}")

输出示例：

标签: electronics, 数量: 3
标签: computer accessories, 数量: 3

详细说明：

• 定义聚合查询：
  • size：设置为0，不返回文档数据。
  • **aggs：**定义一个名为 tags_count 的聚合，使用 terms聚合对 tags.keyword字段进行分组，并返回前10个标签。
    • terms 聚合：将文档按照指定字段的值进行分组，并统计每个组的文档数量。
    • field：指定用于分组的字段，这里使用 tags.keyword，因为 tags 字段是 keyword 类型，适用于精确匹配和分组。
    • size：限制返回的分组数量，这里设置为10，表示返回前10个最常见的标签。
• 执行聚合查询：调用 es.search 方法，传入索引名称和聚合查询，执行聚合操作。
• 处理结果：遍历聚合结果中的每个桶（bucket），打印出标签名称和对应的文档数量。

15.8 更新文档

使用 update API 更新文档中的字段。

# 定义更新内容，更新文档中的 'price' 字段为29.99
update_body = {
    "doc": {
        "price": 29.99
    }
}

# 执行更新操作，指定索引名称、文档ID和更新内容
res = es.update(index=index_name, id=1, body=update_body)

# 输出操作结果，查看文档是否成功更新
print(res['result'])  # 可能的输出包括 'updated'

输出示例：

updated

详细说明：

• 定义更新内容：创建一个字典 update_body，包含要更新的字段和新值。这里，我们将 price 字段的值更新为 29.99。
• 执行更新操作：
  • 使用 es.update 方法，传入索引名称 products、文档ID 1 和更新内容 update_body，执行更新操作。
  • 如果文档存在，price 字段将被更新；如果文档不存在，则会抛出异常。
• 输出结果：通过打印 res['result']，可以看到操作结果，通常为 'updated' 表示文档已成功更新。

15.9 删除文档

使用 delete API 删除指定的文档。

# 定义要删除的文档ID
doc_id = 3

# 执行删除操作，指定索引名称和文档ID
res = es.delete(index=index_name, id=doc_id)

# 输出操作结果，查看文档是否成功删除
print(res['result'])  # 可能的输出包括 'deleted'

输出示例：

deleted

详细说明：

• 定义文档ID：指定要删除的文档的唯一标识符，这里为 3。
• 执行删除操作：
  • 使用 es.delete 方法，传入索引名称 products 和文档ID 3，执行删除操作。
  • 如果文档存在，Elasticsearch 将其删除；如果文档不存在，则会抛出异常。
• 输出结果：通过打印 res['result']，可以看到操作结果，通常为 'deleted' 表示文档已成功删除。

15.10 删除索引

如果需要删除整个索引，可以使用 indices.delete 方法。

# 定义要删除的索引名称
index_to_delete = 'products'

# 执行删除索引操作，忽略404错误（即索引不存在时不抛出异常）
res = es.indices.delete(index=index_to_delete, ignore=[400, 404])

# 输出操作结果，查看索引是否成功删除
print(f"索引 '{index_to_delete}' 删除结果: {res}")

输出示例：

索引 'products' 删除结果: {'acknowledged': True}

详细说明：

• 定义索引名称：指定要删除的索引名称，这里为 products。
• 执行删除索引操作：
  • 使用 es.indices.delete 方法，传入索引名称和 ignore 参数。
  • ignore=[400, 404]：忽略 HTTP 状态码 400（Bad Request）和 404（Not Found）错误，避免因索引不存在而抛出异常。
• 输出结果：通过打印结果字典，查看索引是否成功删除。'acknowledged': True 表示删除请求已被接受和处理。

15.11 实战案例：构建简单的搜索应用

下面我们通过一个简单的 Python 脚本，构建一个命令行搜索应用，用户可以输入关键词，程序会在 Elasticsearch 中搜索并返回匹配的产品。

步骤 1：准备数据

确保 Elasticsearch 中已经有 products 索引，并且已经插入了一些文档。如前面的示例所示。

步骤 2：编写搜索脚本

创建一个名为 search_app.py 的 Python 脚本，并编写以下代码：

import sys
from elasticsearch import Elasticsearch

def search_products(es, index, keyword):
    """
    在指定的索引中搜索包含关键词的产品。

    参数：
    - es: Elasticsearch 客户端实例
    - index: 要搜索的索引名称
    - keyword: 搜索关键词

    返回：
    - 匹配的文档列表
    """
    # 定义多字段匹配查询，搜索 'name' 和 'description' 字段
    query = {
        "query": {
            "multi_match": {
                "query": keyword,              # 用户输入的搜索关键词
                "fields": ["name", "description"]  # 搜索的字段
            }
        }
    }
    
    # 执行搜索操作
    res = es.search(index=index, body=query)
    
    # 返回匹配的文档
    return res['hits']['hits']

def main():
    """
    主函数，处理用户输入并执行搜索。
    """
    # 连接到 Elasticsearch
    es = Elasticsearch([{'host': 'localhost', 'port': 9200}])
    
    # 检查连接是否成功
    if not es.ping():
        print("无法连接到 Elasticsearch")
        sys.exit(1)
    
    index_name = 'products'  # 要搜索的索引名称
    
    # 获取用户输入的关键词
    if len(sys.argv) < 2:
        print("请提供搜索关键词")
        print("使用方法: python search_app.py <关键词>")
        sys.exit(1)
    
    # 将命令行参数拼接成搜索关键词
    keyword = ' '.join(sys.argv[1:])
    
    # 执行搜索
    results = search_products(es, index_name, keyword)
    
    # 显示搜索结果
    print(f"搜索关键词: '{keyword}'")
    print(f"找到 {len(results)} 个匹配的产品")
    for hit in results:
        source = hit["_source"]  # 获取文档的内容
        print(f"ID: {source['product_id']}, 名称: {source['name']}, 价格: {source['price']}, 可用: {source['available']}")
    
if __name__ == "__main__":
    main()

详细说明：

• 导入模块：
  • sys：用于处理命令行参数和退出程序。
  • Elasticsearch：从 elasticsearch 库中导入，用于与 Elasticsearch 交互。
• 定义 search_products 函数：
  • 接受 Elasticsearch 客户端实例、索引名称和搜索关键词作为参数。
  • 使用 multi_match 查询在 name 和 description 字段中搜索关键词，适用于多个字段的全文搜索。
  • 返回匹配的文档列表。
• 定义 main 函数：
  • 连接到 Elasticsearch，并检查连接是否成功。
  • 获取用户在命令行中输入的搜索关键词。如果没有提供关键词，提示用户并退出程序。
  • 调用 search_products 函数执行搜索，并获取结果。
  • 打印搜索关键词和找到的匹配产品数量。
  • 遍历每个匹配的文档，打印产品的ID、名称、价格和可用性。
• 运行脚本：
  • 当脚本作为主程序运行时，调用 main 函数。

步骤 3：运行搜索脚本

在终端中运行脚本，并提供搜索关键词。例如，搜索 “mouse”。

python search_app.py mouse

输出示例：

搜索关键词: 'mouse'
找到 1 个匹配的产品
ID: 1, 名称: Wireless Mouse, 价格: 25.99, 可用: True

详细说明：

• 执行命令：在终端中运行 python search_app.py mouse，其中 mouse 是用户输入的搜索关键词。
• 输出结果：
  • 显示搜索关键词。
  • 显示找到的匹配产品数量。
  • 列出每个匹配产品的详细信息，包括ID、名称、价格和是否可用。

15.12 错误处理与调试

在实际开发中，可能会遇到各种错误。以下是一些常见的错误处理方法。

15.12.1 连接错误

确保 Elasticsearch 服务正在运行，并且客户端配置正确。

from elasticsearch import Elasticsearch, ConnectionError

try:
    # 尝试连接到 Elasticsearch
    es = Elasticsearch([{'host': 'localhost', 'port': 9200}])
    es.ping()  # 检查连接
except ConnectionError as e:
    # 如果连接失败，捕获异常并打印错误信息
    print("无法连接到 Elasticsearch:", e)
    sys.exit(1)

详细说明：

• 导入异常类：从 elasticsearch 库中导入 ConnectionError，用于捕获连接相关的异常。
• 尝试连接：使用 try-except 块尝试连接到 Elasticsearch，并使用 ping() 方法检查连接是否成功。
• 捕获异常：如果连接失败，捕获 ConnectionError 异常，打印错误信息，并通过 sys.exit(1) 退出程序。

15.12.2 索引不存在

在执行操作前，检查索引是否存在。

index_name = 'products'

# 检查索引是否存在
if not es.indices.exists(index=index_name):
    print(f"索引 '{index_name}' 不存在")
    sys.exit(1)
else:
    print(f"索引 '{index_name}' 存在")
    # 继续执行后续操作

详细说明：

• 定义索引名称：指定要检查的索引名称，这里为 products。
• 检查索引存在性：使用 es.indices.exists 方法检查索引是否存在。
• 处理结果：
  • 如果索引不存在，打印提示信息并退出程序。
  • 如果索引存在，打印确认信息，并继续执行后续操作。

15.12.3 文档不存在

在更新或删除文档前，检查文档是否存在。

doc_id = 1
index_name = 'products'

# 检查文档是否存在
if not es.exists(index=index_name, id=doc_id):
    print(f"文档 ID '{doc_id}' 不存在")
else:
    # 如果文档存在，可以执行更新或删除操作
    # 示例：删除文档
    res = es.delete(index=index_name, id=doc_id)
    print(f"文档 ID '{doc_id}' 删除结果: {res['result']}")

详细说明：

• 定义文档ID和索引名称：指定要检查的文档ID和索引名称。
• 检查文档存在性：使用 es.exists 方法检查指定文档是否存在。
• 处理结果：
  • 如果文档不存在，打印提示信息。
  • 如果文档存在，可以继续执行更新或删除操作。这里以删除操作为例，调用 es.delete 方法删除文档，并打印操作结果。

15.13 使用高级功能

15.13.1 自动完成（Autocomplete）

实现输入建议功能，提升用户搜索体验。

# 定义索引映射时添加 suggest 字段
index_body = {
    "settings": {
        "number_of_shards": 3,    # 主分片数量
        "number_of_replicas": 1   # 副本分片数量
    },
    "mappings": {
        "properties": {
            "product_id": { "type": "long" },
            "name": { "type": "text" },
            "description": { "type": "text" },
            "price": { "type": "double" },
            "available": { "type": "boolean" },
            "tags": { "type": "keyword" },
            "created_at": { "type": "date" },
            "suggest": { "type": "completion" }    # 定义 suggest 字段为 completion 类型
        }
    }
}

# 创建索引时使用上述映射
if not es.indices.exists(index=index_name):
    es.indices.create(index=index_name, body=index_body)
    print(f"索引 '{index_name}' 创建成功")
else:
    print(f"索引 '{index_name}' 已存在")

# 插入带有 suggest 字段的文档
doc = {
    "product_id": 1,
    "name": "Wireless Mouse",
    "description": "A comfortable wireless mouse with ergonomic design.",
    "price": 25.99,
    "available": True,
    "tags": ["electronics", "computer accessories"],
    "created_at": "2023-01-15",
    "suggest": {
        "input": ["Wireless Mouse", "Mouse", "Wireless"]  # 定义自动完成的输入选项
    }
}

# 插入文档
es.index(index=index_name, id=1, document=doc)
print("文档插入完成，包含 suggest 字段")

# 定义建议查询，使用前缀 'wir' 进行自动完成
suggest_query = {
    "suggest": {
        "product-suggest": {
            "prefix": "wir",               # 用户输入的前缀
            "completion": {
                "field": "suggest",        # 使用 'suggest' 字段进行自动完成
                "fuzzy": {
                    "fuzziness": 2          # 允许的拼写错误（模糊匹配）
                }
            }
        }
    }
}

# 执行建议查询
res = es.search(index=index_name, body=suggest_query)

# 处理建议结果，遍历每个建议选项并打印
suggestions = res['suggest']['product-suggest'][0]['options']
for option in suggestions:
    print(option['text'])  # 打印建议的文本

输出示例：

文档插入完成，包含 suggest 字段
Wireless Mouse
Wireless

详细说明：

• 定义索引映射时添加 suggest 字段：
  • 在 mappings 中添加一个名为 suggest 的字段，类型为 completion，用于支持自动完成（Suggesters）功能。
• 创建索引：
  • 如果索引不存在，使用上述映射创建索引。
• 插入文档：
  • 在文档中添加 suggest 字段，并定义多个输入选项（input），如 “Wireless Mouse”、“Mouse” 和 “Wireless”。这些选项将用于自动完成建议。
• 定义建议查询：
  • 使用 suggest 部分定义建议查询。
  • prefix：用户输入的前缀，这里为 “wir”。
  • completion：指定使用的 suggest 字段，并设置 fuzziness 为2，允许拼写错误。
• 执行建议查询：调用 es.search 方法，传入索引名称和建议查询，执行自动完成查询。
• 处理结果：
  • 从响应中提取建议选项列表。
  • 遍历每个建议选项，打印建议的文本内容。

15.13.2 地理位置搜索（Geo Search）

基于地理位置的数据搜索，适用于位置相关的应用场景。

# 定义索引映射时添加 location 字段
index_body = {
    "settings": {
        "number_of_shards": 3,    # 主分片数量
        "number_of_replicas": 1   # 副本分片数量
    },
    "mappings": {
        "properties": {
            "product_id": { "type": "long" },
            "name": { "type": "text" },
            "description": { "type": "text" },
            "price": { "type": "double" },
            "available": { "type": "boolean" },
            "tags": { "type": "keyword" },
            "created_at": { "type": "date" },
            "location": { "type": "geo_point" }   # 定义 location 字段为 geo_point 类型
        }
    }
}

# 创建索引时使用上述映射
if not es.indices.exists(index=index_name):
    es.indices.create(index=index_name, body=index_body)
    print(f"索引 '{index_name}' 创建成功")
else:
    print(f"索引 '{index_name}' 已存在")

# 插入带有 location 字段的文档
doc = {
    "product_id": 1,
    "name": "Wireless Mouse",
    "description": "A comfortable wireless mouse with ergonomic design.",
    "price": 25.99,
    "available": True,
    "tags": ["electronics", "computer accessories"],
    "created_at": "2023-01-15",
    "location": { "lat": 40.7128, "lon": -74.0060 }  # 定义地理位置坐标（纬度和经度）
}

# 插入文档
es.index(index=index_name, id=1, document=doc)
print("文档插入完成，包含 location 字段")

# 定义地理位置搜索查询，查找距离指定点10公里以内的产品
geo_query = {
    "query": {
        "bool": {
            "filter": {
                "geo_distance": {
                    "distance": "10km",                       # 搜索半径为10公里
                    "location": { "lat": 40.7128, "lon": -74.0060 }  # 指定中心点的坐标
                }
            }
        }
    }
}

# 执行地理位置搜索查询
res = es.search(index=index_name, body=geo_query)

# 处理搜索结果，遍历每个匹配的文档并打印
print(f"找到 {res['hits']['total']['value']} 个匹配的产品")
for hit in res['hits']['hits']:
    source = hit["_source"]  # 获取文档内容
    print(f"ID: {source['product_id']}, 名称: {source['name']}, 位置: {source['location']}")

输出示例：

文档插入完成，包含 location 字段
找到 1 个匹配的产品
ID: 1, 名称: Wireless Mouse, 位置: {'lat': 40.7128, 'lon': -74.006}

详细说明：

• 定义索引映射时添加 location 字段：
  • 在 mappings 中添加一个名为 location 的字段，类型为 geo_point，用于存储地理位置坐标（纬度和经度）。
• 创建索引：
  • 如果索引不存在，使用上述映射创建索引。
• 插入文档：
  • 在文档中添加 location 字段，并定义具体的地理坐标，例如纽约市的坐标（纬度40.7128，经度-74.0060）。
• 定义地理位置搜索查询：
  • 使用 geo_distance 查询，查找距离指定地理位置10公里以内的产品。
  • distance：指定搜索半径，这里为10公里。
  • location：指定中心点的坐标。
• 执行地理位置搜索查询：调用 es.search 方法，传入索引名称和地理位置查询，执行搜索操作。
• 处理结果：
  • 打印匹配的产品数量。
  • 遍历每个匹配的文档，打印产品的ID、名称和地理位置坐标。