MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

ElasticSearch API缩略处理的作用与实现

2022-08-104.4k 阅读

ElasticSearch API 缩略处理的作用

提升开发效率

在使用 ElasticSearch 进行应用开发时，原生的 API 往往具有较为冗长和复杂的调用方式。例如，在进行简单的文档索引操作时，原生 API 可能需要开发者详细指定多个参数，包括索引名称、类型（在 ElasticSearch 7.x 及之后版本，类型的概念逐渐弱化）、文档 ID 以及文档内容等。这对于开发人员来说，每次编写这样的代码都需要花费一定的时间和精力来确保参数的准确性。

而通过 API 缩略处理，可以将这些常见的操作封装成更为简洁的方法。以 Java 客户端为例，假设我们使用原生的 Elasticsearch Java High - Level REST Client 进行文档索引：

RestHighLevelClient client = new RestHighLevelClient(
        RestClient.builder(
                new HttpHost("localhost", 9200, "http")));

IndexRequest indexRequest = new IndexRequest("my_index")
      .id("1")
      .source(XContentType.JSON, "field1", "value1", "field2", "value2");

try {
    IndexResponse indexResponse = client.index(indexRequest, RequestOptions.DEFAULT);
} catch (IOException e) {
    e.printStackTrace();
}

经过 API 缩略处理后，我们可以封装一个简单的方法：

public void simpleIndex(String index, String id, Map<String, Object> source) {
    IndexRequest indexRequest = new IndexRequest(index)
          .id(id)
          .source(source);
    try {
        IndexResponse indexResponse = client.index(indexRequest, RequestOptions.DEFAULT);
    } catch (IOException e) {
        e.printStackTrace();
    }
}

然后调用该方法：

Map<String, Object> doc = new HashMap<>();
doc.put("field1", "value1");
doc.put("field2", "value2");
simpleIndex("my_index", "1", doc);

这样开发人员在进行索引操作时，代码更加简洁明了，减少了重复代码的编写，从而大大提升了开发效率。

降低学习成本

对于刚接触 ElasticSearch 的开发人员来说，其庞大而复杂的 API 体系是一个不小的学习障碍。原生 API 涵盖了从基本的文档操作（增、删、改、查）到复杂的聚合分析、分布式管理等众多功能，每个功能又有多种参数组合和调用方式。例如，在进行搜索操作时，不仅要了解查询 DSL（Domain - Specific Language）的语法，还要掌握如何在 API 中正确构建查询请求对象，包括设置查询条件、排序、分页等参数。

API 缩略处理可以将常用的功能以一种更直观、更易于理解的方式呈现给开发人员。比如，我们可以封装一个简单的搜索方法，只需要传入索引名称、查询字符串和分页参数即可实现基本的搜索功能：

public SearchResponse simpleSearch(String index, String query, int from, int size) {
    SearchRequest searchRequest = new SearchRequest(index);
    SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
    sourceBuilder.query(QueryBuilders.queryStringQuery(query));
    sourceBuilder.from(from);
    sourceBuilder.size(size);
    searchRequest.source(sourceBuilder);
    try {
        return client.search(searchRequest, RequestOptions.DEFAULT);
    } catch (IOException e) {
        e.printStackTrace();
    }
    return null;
}

调用时：

SearchResponse response = simpleSearch("my_index", "field1:value1", 0, 10);

通过这种方式，新上手的开发人员不需要一开始就深入研究 ElasticSearch 复杂的 API 细节，只需要掌握这些简化后的接口，就能快速开始进行开发工作，降低了学习曲线，使开发人员能够更快地将 ElasticSearch 集成到项目中。

提高代码可读性和可维护性

在大型项目中，代码的可读性和可维护性至关重要。当使用原生 ElasticSearch API 时，代码中可能会充斥着大量的 API 调用细节，使得代码逻辑变得不清晰。例如，在进行复杂的聚合分析时，原生 API 可能需要构建多层嵌套的对象来定义聚合条件、子聚合等。

SearchRequest searchRequest = new SearchRequest("my_index");
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();

TermsAggregationBuilder termsAggregationBuilder = AggregationBuilders.terms("by_field1")
      .field("field1");

AvgAggregationBuilder avgAggregationBuilder = AggregationBuilders.avg("avg_field2")
      .field("field2");

termsAggregationBuilder.subAggregation(avgAggregationBuilder);

sourceBuilder.aggregation(termsAggregationBuilder);

searchRequest.source(sourceBuilder);

try {
    SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
    Aggregations aggregations = searchResponse.getAggregations();
    ParsedStringTerms byField1 = aggregations.get("by_field1");
    for (Terms.Bucket bucket : byField1.getBuckets()) {
        ParsedAvg avgField2 = bucket.getAggregations().get("avg_field2");
        System.out.println("Field1 value: " + bucket.getKeyAsString() + ", Avg of field2: " + avgField2.getValue());
    }
} catch (IOException e) {
    e.printStackTrace();
}

通过 API 缩略处理，我们可以将这些复杂的聚合操作封装成一个方法：

public void complexAggregation(String index) {
    SearchRequest searchRequest = new SearchRequest(index);
    SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();

    TermsAggregationBuilder termsAggregationBuilder = AggregationBuilders.terms("by_field1")
          .field("field1");

    AvgAggregationBuilder avgAggregationBuilder = AggregationBuilders.avg("avg_field2")
          .field("field2");

    termsAggregationBuilder.subAggregation(avgAggregationBuilder);

    sourceBuilder.aggregation(termsAggregationBuilder);

    searchRequest.source(sourceBuilder);

    try {
        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
        Aggregations aggregations = searchResponse.getAggregations();
        ParsedStringTerms byField1 = aggregations.get("by_field1");
        for (Terms.Bucket bucket : byField1.getBuckets()) {
            ParsedAvg avgField2 = bucket.getAggregations().get("avg_field2");
            System.out.println("Field1 value: " + bucket.getKeyAsString() + ", Avg of field2: " + avgField2.getValue());
        }
    } catch (IOException e) {
        e.printStackTrace();
    }
}

调用时：

complexAggregation("my_index");

这样代码的逻辑更加清晰，其他开发人员在阅读和维护代码时，能够更快地理解代码的功能，而不需要花费大量时间去解析复杂的 API 调用细节。同时，如果 ElasticSearch API 发生变化，只需要在封装的方法内部进行修改，而不需要在整个项目中到处查找和修改相关的 API 调用代码，提高了代码的可维护性。

ElasticSearch API 缩略处理的实现

基于客户端库的封装

Java 客户端封装

  1. Maven 依赖 首先，确保项目中引入了 Elasticsearch Java High - Level REST Client 的依赖。在 pom.xml 文件中添加如下依赖：
<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch - rest - high - level - client</artifactId>
    <version>7.17.0</version>
</dependency>
<dependency>
    <groupId>org.elasticsearch</groupId>
    <artifactId>elasticsearch</artifactId>
    <version>7.17.0</version>
</dependency>
  1. 封装索引方法 如前文所述，我们可以封装一个简单的索引方法。创建一个 ElasticsearchUtil 类：
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.common.xcontent.XContentType;

import java.io.IOException;
import java.util.Map;

public class ElasticsearchUtil {
    private static RestHighLevelClient client;

    static {
        client = new RestHighLevelClient(
                RestClient.builder(
                        new HttpHost("localhost", 9200, "http")));
    }

    public static void simpleIndex(String index, String id, Map<String, Object> source) {
        IndexRequest indexRequest = new IndexRequest(index)
              .id(id)
              .source(source);
        try {
            IndexResponse indexResponse = client.index(indexRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
  1. 封装搜索方法 接着，我们可以封装搜索方法：
import org.apache.http.HttpHost;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;

import java.io.IOException;

public class ElasticsearchUtil {
    private static RestHighLevelClient client;

    static {
        client = new RestHighLevelClient(
                RestClient.builder(
                        new HttpHost("localhost", 9200, "http")));
    }

    public static SearchResponse simpleSearch(String index, String query, int from, int size) {
        SearchRequest searchRequest = new SearchRequest(index);
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(QueryBuilders.queryStringQuery(query));
        sourceBuilder.from(from);
        sourceBuilder.size(size);
        searchRequest.source(sourceBuilder);
        try {
            return client.search(searchRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
}

Python 客户端封装

  1. 安装 Elasticsearch 库 使用 pip 安装 Elasticsearch 库：
pip install elasticsearch
  1. 封装索引方法 创建一个 es_utils.py 文件，编写如下代码：
from elasticsearch import Elasticsearch

es = Elasticsearch([{'host': 'localhost', 'port': 9200}])


def simple_index(index, id, body):
    try:
        es.index(index=index, id=id, body=body)
    except Exception as e:
        print(f"Error indexing document: {e}")
  1. 封装搜索方法
from elasticsearch import Elasticsearch
from elasticsearch_dsl import Search

es = Elasticsearch([{'host': 'localhost', 'port': 9200}])


def simple_search(index, query, from_num, size):
    s = Search(using=es, index=index) \
      .query('query_string', query=query) \
      .extra(from_=from_num, size=size)
    try:
        response = s.execute()
        return response
    except Exception as e:
        print(f"Error searching: {e}")
        return None

自定义中间件实现

原理

自定义中间件可以在应用程序与 ElasticSearch 之间起到桥梁的作用。它接收应用程序发送的简化请求，然后将其转换为 ElasticSearch 原生 API 能够理解的请求格式。例如，应用程序可能发送一个简单的包含索引名称、文档 ID 和文档内容的请求来进行索引操作，中间件接收到这个请求后，根据其内部的逻辑将其转换为 ElasticSearch 的 IndexRequest 对象，并调用 ElasticSearch 客户端进行实际的索引操作。

同时，中间件还可以对 ElasticSearch 返回的响应进行处理，将其转换为应用程序能够理解的更简洁的格式。比如，ElasticSearch 的搜索响应可能包含大量的元数据和详细的搜索结果结构，中间件可以提取出应用程序关心的部分，如文档列表、命中总数等，返回给应用程序。

实现示例（以 Node.js 为例）

  1. 安装依赖 在项目目录下，使用 npm 安装 elasticsearch 库：
npm install elasticsearch
  1. 创建中间件 创建一个 esMiddleware.js 文件：
const { Client } = require('elasticsearch');

const esClient = new Client({
    node: 'http://localhost:9200'
});

function simpleIndex(index, id, body) {
    return new Promise((resolve, reject) => {
        esClient.index({
            index: index,
            id: id,
            body: body
        }, (error, response) => {
            if (error) {
                reject(error);
            } else {
                resolve(response);
            }
        });
    });
}

function simpleSearch(index, query, from, size) {
    return new Promise((resolve, reject) => {
        esClient.search({
            index: index,
            body: {
                query: {
                    query_string: {
                        query: query
                    }
                },
                from: from,
                size: size
            }
        }, (error, response) => {
            if (error) {
                reject(error);
            } else {
                const hits = response.hits.hits.map(hit => hit._source);
                const total = response.hits.total.value;
                resolve({ hits, total });
            }
        });
    });
}

module.exports = {
    simpleIndex,
    simpleSearch
};
  1. 在应用程序中使用中间件 假设我们有一个简单的 Node.js 应用程序，使用 Express 框架：
const express = require('express');
const { simpleIndex, simpleSearch } = require('./esMiddleware');

const app = express();
app.use(express.json());

app.post('/index', async (req, res) => {
    try {
        const { index, id, body } = req.body;
        await simpleIndex(index, id, body);
        res.status(200).send('Document indexed successfully');
    } catch (error) {
        res.status(500).send(`Error indexing document: ${error.message}`);
    }
});

app.get('/search', async (req, res) => {
    try {
        const { index, query, from = 0, size = 10 } = req.query;
        const result = await simpleSearch(index, query, from, size);
        res.status(200).send(result);
    } catch (error) {
        res.status(500).send(`Error searching: ${error.message}`);
    }
});

const port = 3000;
app.listen(port, () => {
    console.log(`Server running on port ${port}`);
});

基于框架扩展的实现

Spring Data Elasticsearch 扩展

  1. Maven 依赖 在 Spring Boot 项目的 pom.xml 文件中添加 Spring Data Elasticsearch 依赖：
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring - boot - starter - data - elasticsearch</artifactId>
</dependency>
  1. 自定义 Repository 方法 假设我们有一个 Book 实体类：
import org.springframework.data.elasticsearch.annotations.Document;

@Document(indexName = "books")
public class Book {
    private String id;
    private String title;
    private String author;

    // getters and setters
}

创建一个 BookRepository 接口，继承自 ElasticsearchRepository：

import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;

public interface BookRepository extends ElasticsearchRepository<Book, String> {
    // 自定义方法，简化搜索
    Iterable<Book> findByTitleContaining(String title);
}
  1. 使用自定义方法 在服务层中使用该方法：
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.stream.Collectors;

@Service
public class BookService {
    @Autowired
    private BookRepository bookRepository;

    public List<Book> searchBooksByTitle(String title) {
        return bookRepository.findByTitleContaining(title)
              .stream()
              .collect(Collectors.toList());
    }
}

Django Elasticsearch DSL 扩展

  1. 安装依赖 在 Django 项目环境中，使用 pip 安装 django - elasticsearch - dsl：
pip install django - elasticsearch - dsl
  1. 定义 Document 和 Index 在 Django 应用的 documents.py 文件中定义：
from django_elasticsearch_dsl import Document, fields
from django_elasticsearch_dsl.registries import registry
from.models import Book

@registry.register_document
class BookDocument(Document):
    title = fields.TextField()
    author = fields.TextField()

    class Index:
        name = 'books'
        settings = {
            'number_of_shards': 1,
            'number_of_replicas': 0
        }

    class Django:
        model = Book
        fields = [
            'id'
        ]
  1. 自定义搜索方法 在视图函数中可以自定义搜索逻辑：
from django.http import JsonResponse
from.models import Book
from elasticsearch_dsl import Search
from django_elasticsearch_dsl.registries import registry

def search_books(request):
    query = request.GET.get('query', '')
    es = registry.get_connection()
    s = Search(using=es, index='books') \
      .query('match', title=query)
    response = s.execute()
    results = [{'title': hit.title, 'author': hit.author} for hit in response]
    return JsonResponse({'results': results})

通过以上几种方式，我们可以有效地实现 ElasticSearch API 的缩略处理，充分发挥其在提升开发效率、降低学习成本以及提高代码质量方面的作用。无论是基于客户端库的直接封装，还是通过自定义中间件或框架扩展的方式，都为开发人员提供了更便捷、高效地使用 ElasticSearch 的途径。在实际项目中，开发人员可以根据项目的规模、技术栈以及具体需求来选择合适的实现方式。例如，对于小型项目或对灵活性要求较高的项目，基于客户端库的封装可能更为合适；而对于大型企业级项目，借助框架扩展或自定义中间件可以更好地与现有系统架构进行整合，实现更全面的功能定制和管理。同时，随着 ElasticSearch 版本的不断更新和功能的不断增强，开发人员需要持续关注并适时调整 API 缩略处理的实现，以确保应用程序能够始终保持高效、稳定的运行。