elasticsearch使用指南之Elasticsearch Document Multi Get API、Bulk API详解、原理与示例

本文涉及的产品
检索分析服务 Elasticsearch 版,2核4GB开发者规格 1个月
简介: 本文详细介绍如何elasticsearch Multi Get API与Bulk API,并介绍其基本运行原理,最后给出各个API的java示例。

作者简介:《RocketMQ技术内幕》作者、中间件兴趣圈微信公众号维护者。

本文将详细介绍elasticsearch批量获取API(Multi Get API)和Bulk API的使用。

1、Multi Get API
详细API如下:

  • public final MultiGetResponse mget(MultiGetRequest multiGetRequest, RequestOptions options) throws IOException
  • public final void mgetAsync(MultiGetRequest multiGetRequest, RequestOptions options, ActionListener listener)

其核心需要关注MultiGetRequest 。
clipboard

从上面所知,mget及批量获取文档,通过add方法添加多个Item,每一个item代表一个文件获取请求,其相关字段已在get API中详细介绍,这里就不做过多详解。

Mget API使用示例

public static void testMget() {
        RestHighLevelClient client = EsClient.getClient();
        try {
            MultiGetRequest request = new MultiGetRequest();
            request.add("twitter", "_doc", "10");
            request.add("twitter", "_doc", "11");
            request.add("twitter", "_doc", "12");
            request.add("gisdemo", "_doc", "10");
            MultiGetResponse result = client.mget(request, RequestOptions.DEFAULT);
            System.out.println(result);
        } catch (Throwable e) {
            e.printStackTrace();
        } finally {
            EsClient.close(client);
        }
    }

返回的结果其本质是一个 GetResponse的数组,不会因为其中一个失败,整个请求失败,但其结果中会标明每一个是否成功。

其返回结果类图如下:
clipboard

其字段过滤(Source filtering)、路由等机制与Get API相同,详情请参考:Elasticsearch Document Get API详解、原理与示例

2、Elasticsearch Bulk API

Bulk API可以在一次API调用中包含多个索引操作,例如更新索引,删除索引等,类比批量操作。
详细API如下:

  • public final BulkResponse bulk(BulkRequest bulkRequest, RequestOptions options) throws IOException
  • public final void bulkAsync(BulkRequest bulkRequest, RequestOptions options, ActionListener listener)

2.1 BulkRequest详解
clipboard

我们先一一来看一下其核心属性与与典型方法:

  • final List requests = new ArrayList<>():单个命令容器,DocWriteRequest的子类包括:IndexRequest、UpdateRequest、DeleteRequest。
  • private final Set indices = new HashSet<>():List requests涉及到的索引。
    List
  • protected TimeValue timeout = BulkShardRequest.DEFAULT_TIMEOUT:timeout机制,针对一个Bulk请求生效。
  • private ActiveShardCount waitForActiveShards = ActiveShardCount.DEFAULT: waitForActiveShards,针对一个Bulk请求生效,各个请求中waitForActiveShards优先。
  • private RefreshPolicy refreshPolicy = RefreshPolicy.NONE:刷新策略。
  • private long sizeInBytes = 0:整个Bulk请求的大小。

通过add api为BulkRequest添加一个请求。

2.2 Bulk API请求格式详解
Bulk Rest请求协议基于如下格式:

POST _bulk
{ "index" : { "_index" : "test", "_type" : "_doc", "_id" : "1" } }
{ "field1" : "value1" }
{ "delete" : { "_index" : "test", "_type" : "_doc", "_id" : "2" } }
{ "create" : { "_index" : "test", "_type" : "_doc", "_id" : "3" } }
{ "field1" : "value3" }
{ "update" : {"_id" : "1", "_type" : "_doc", "_index" : "test"} }
{ "doc" : {"field2" : "value2"} }

请求格式如下(restfull):

  • POST请求,其Content-Type为application/x-ndjson。
  • 每一个命令占用两行,每行的结束字符为rn。
  • 第一行为元数据,"opType" : {元数据}。
  • 第二行为有效载体(非必选),例如Index操作,其有效载荷为IndexRequest#source字段。
  • opType可选值 index、create、update、delete。

公用元数据(index、create、update、delete)如下
1)_index :索引名
2)_type:类型名
3)_id:文档ID
4)routing:路由值
5)parent
6)version:数据版本号
7)version_type:版本类型
各操作特有元数据
1、index | create
1)pipeline
2、update
1)retry_on_conflict :更新冲突时重试次数。
2)_source:字段过滤。

有效载荷说明
1、index | create
其有效载荷为_source字段。
2、update
其有效载荷为:partial doc, upsert and script。
3、delete
没有有效载荷。

请求格式为什么要设计成metdata+有效载体的方式,主要是为了在接收端节点(所谓的接收端节点是指收到命令的第一节点),只需解析 metadata,然后将请求直接转发给对应的数据节点。

2.3 bulk API通用特性分析
2.3.1 版本管理
每一个Bulk条目拥有独自的version,存在于请求条目的item的元数据中。

2.3.2 路由
每一个Bulk条目各自生效。

2.3.3 Wait For Active Shards
通常可以设置BulkRequest#waitForActiveShards来要求Bulk批量执行之前要求处于激活的最小副本数。

2.3.4 Bulk Demo

public static final void testBulk() {
        RestHighLevelClient client = EsClient.getClient();
        try {
            IndexRequest indexRequest = new IndexRequest("twitter", "_doc", "12")
                    .source(buildTwitter("dingw", "2009-11-18T14:12:12", "test bulk"));
            UpdateRequest updateRequest = new UpdateRequest("twitter", "_doc", "11")
                        .doc(new IndexRequest("twitter", "_doc", "11")
                                .source(buildTwitter("dingw", "2009-11-18T14:12:12", "test bulk update")));
            BulkRequest request = new BulkRequest();
            request.add(indexRequest);
            request.add(updateRequest);
            BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);
            for (BulkItemResponse bulkItemResponse : bulkResponse) { 
                if (bulkItemResponse.isFailed()) { 
                    BulkItemResponse.Failure failure = bulkItemResponse.getFailure(); 
                    System.out.println(failure);
                    continue;
                }
                DocWriteResponse itemResponse = bulkItemResponse.getResponse(); 
                if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.INDEX
                        || bulkItemResponse.getOpType() == DocWriteRequest.OpType.CREATE) { 
                    IndexResponse indexResponse = (IndexResponse) itemResponse;
                    System.out.println(indexRequest);
                } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.UPDATE) {
                    UpdateResponse updateResponse = (UpdateResponse) itemResponse;
                    System.out.println(updateRequest);
                } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.DELETE) { 
                    DeleteResponse deleteResponse = (DeleteResponse) itemResponse;
                    System.out.println(deleteResponse);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            EsClient.close(client);
        }
    }

批量更新bulk api就介绍到这里了。

相关实践学习
使用阿里云Elasticsearch体验信息检索加速
通过创建登录阿里云Elasticsearch集群,使用DataWorks将MySQL数据同步至Elasticsearch,体验多条件检索效果,简单展示数据同步和信息检索加速的过程和操作。
ElasticSearch 入门精讲
ElasticSearch是一个开源的、基于Lucene的、分布式、高扩展、高实时的搜索与数据分析引擎。根据DB-Engines的排名显示,Elasticsearch是最受欢迎的企业搜索引擎,其次是Apache Solr(也是基于Lucene)。 ElasticSearch的实现原理主要分为以下几个步骤: 用户将数据提交到Elastic Search 数据库中 通过分词控制器去将对应的语句分词,将其权重和分词结果一并存入数据 当用户搜索数据时候,再根据权重将结果排名、打分 将返回结果呈现给用户 Elasticsearch可以用于搜索各种文档。它提供可扩展的搜索,具有接近实时的搜索,并支持多租户。
目录
相关文章
|
7月前
|
存储 API 索引
Elasticsearch Reroute API 的使用
Elasticsearch Reroute API 的使用
122 1
|
7月前
|
存储 JSON 自然语言处理
Elasticsearch 利用API进行搜索
Elasticsearch 利用API进行搜索
65 0
|
2月前
|
存储 人工智能 自然语言处理
Elasticsearch Inference API增加对阿里云AI的支持
本文将介绍如何在 Elasticsearch 中设置和使用阿里云的文本生成、重排序、稀疏向量和稠密向量服务,提升搜索相关性。
75 14
Elasticsearch Inference API增加对阿里云AI的支持
|
4月前
|
API 开发者 Python
API接口:原理、实现及应用
本文详细介绍了API接口在现代软件开发中的重要性及其工作原理。API接口作为应用程序间通信的桥梁,通过预定义的方法和协议实现数据和服务的共享。文章首先解释了API接口的概念,接着通过Python Flask框架示例展示了API的设计与实现过程,并强调了安全性的重要性。最后,本文还讨论了API接口在Web服务和移动应用程序等领域的广泛应用场景。
|
4月前
|
Kubernetes 负载均衡 API
在K8S中,api-service 和 kube-schedule 高可用原理是什么?
在K8S中,api-service 和 kube-schedule 高可用原理是什么?
|
7月前
|
XML 搜索推荐 API
通义千问API:让大模型使用各种工具
本章我们将通过一个简单的例子,揭示基于LangChain的Agent开发的秘密,从而了解如何扩展大模型的能力。
通义千问API:让大模型使用各种工具
|
6月前
|
存储 缓存 Java
JavaSE 字符串String及相关API StringBuilder StringJoiner 底层原理 详解
JavaSE 字符串String及相关API StringBuilder StringJoiner 底层原理 详解
49 2
|
6月前
|
人工智能 自然语言处理 搜索推荐
Elasticsearch 开放 inference API 增加了对 Azure OpenAI 嵌入的支持
【6月更文挑战第8天】Elasticsearch 推出开放 inference API,支持 Azure OpenAI 嵌入,强化搜索和数据分析能力。此更新使用户能灵活集成 AI 技术,实现智能精准搜索。Azure OpenAI 的语言理解能力优化了用户查询处理,提升搜索相关性。示例代码显示了如何结合两者处理查询。该创新提升数据检索效率,适用于智能客服和推荐系统,但也带来数据安全和模型准确性等挑战。这标志着搜索和数据分析领域的智能化新阶段,期待更多创新应用。未来,我们需要持续探索和完善,以发挥技术的最大潜力。
49 3
|
6月前
|
存储 缓存 Java
掌握Elasticsearch集群参数查询API
掌握Elasticsearch集群参数查询API
|
7月前
|
安全 算法 Java
Java Stream API:原理、应用与深入解析
Java Stream API:原理、应用与深入解析
316 2