Elasticsearch java api(五) Bulk批量索引

一、Bulk API

使用bulk命令时，REST API以_bulk结尾,批量操作写在json文件中，官网给出的语法格式：

action_and_meta_data\n
optional_source\n
action_and_meta_data\n
optional_source\n
....
action_and_meta_data\n
optional_source\n

也就是说每一个操作都有2行数据组成，末尾要回车换行。第一行用来说明操作命令和原数据、第二行是自定义的选项.举个例子，同时执行插入2条数据、删除一条数据, 新建bulkdata.json,写入如下内容：

{ "create" : { "_index" : "blog", "_type" : "article", "_id" : "3" }}
{ "title":"title1","posttime":"2016-07-02","content":"内容一" }{ "create" : { "_index" : "blog", "_type" : "article", "_id" : "4" }}
{ "title":"title2","posttime":"2016-07-03","content":"内容2" }{ "delete":{"_index" : "blog", "_type" : "article", "_id" : "1" }}

执行：

$ curl -XPOST "http://localhost:9200/_bulk?pretty" --data-binary @bulkAdd.json
{"took" : 11,"errors" : false,"items" : [ {"create" : {"_index" : "blog","_type" : "article","_id" : "13","_version" : 1,"_shards" : {"total" : 1,"successful" : 1,"failed" : 0},"status" : 201}} ]
}

注意：行末要回车换行，不然会因为命令不能识别而出现错误.

$ curl -XPOST "http://localhost:9200/_bulk?pretty" --data-binary @bulkAdd.json
{"error" : {"root_cause" : [ {"type" : "action_request_validation_exception","reason" : "Validation Failed: 1: no requests added;"} ],"type" : "action_request_validation_exception","reason" : "Validation Failed: 1: no requests added;"},"status" : 400
}

二、批量导出

下面的例子是把索引库中的文档以json格式批量导出到文件中，其中集群名称为”bropen”,索引库名为”blog”,type为”article”,项目根目录下新建files/bulk.txt,索引内容写入bulk.txt中:

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHits;public class ElasticSearchBulkOut {public static void main(String[] args) {try {Settings settings = Settings.settingsBuilder().put("cluster.name", "bropen").build();// cluster.name在elasticsearch.ymlClient client = TransportClient.builder().settings(settings).build().addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("127.0.0.1"), 9300));QueryBuilder qb = QueryBuilders.matchAllQuery();SearchResponse response = client.prepareSearch("blog").setTypes("article").setQuery(QueryBuilders.matchAllQuery()).execute().actionGet();SearchHits resultHits = response.getHits();File article = new File("files/bulk.txt");FileWriter fw = new FileWriter(article);BufferedWriter bfw = new BufferedWriter(fw);if (resultHits.getHits().length == 0) {System.out.println("查到0条数据!");} else {for (int i = 0; i < resultHits.getHits().length; i++) {String jsonStr = resultHits.getHits()[i].getSourceAsString();System.out.println(jsonStr);bfw.write(jsonStr);bfw.write("\n");}}bfw.close();fw.close();} catch (UnknownHostException e) {e.printStackTrace();} catch (IOException e) {e.printStackTrace();}}}

三、批量导入

从刚才导出的bulk.txt文件中按行读取，然后bulk导入。首先通过调用client.prepareBulk()实例化一个BulkRequestBuilder对象,调用BulkRequestBuilder对象的add方法添加数据。实现代码：

import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;public class ElasticSearchBulkIn {public static void main(String[] args) {try {Settings settings = Settings.settingsBuilder().put("cluster.name", "bropen").build();// cluster.name在elasticsearch.yml中配置Client client = TransportClient.builder().settings(settings).build().addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("127.0.0.1"), 9300));File article = new File("files/bulk.txt");FileReader fr=new FileReader(article);BufferedReader bfr=new BufferedReader(fr);String line=null;BulkRequestBuilder bulkRequest=client.prepareBulk();int count=0;while((line=bfr.readLine())!=null){bulkRequest.add(client.prepareIndex("test","article").setSource(line));if (count%10==0) {bulkRequest.execute().actionGet();}count++;//System.out.println(line);}bulkRequest.execute().actionGet();bfr.close();fr.close();} catch (UnknownHostException e) {e.printStackTrace();} catch (FileNotFoundException e) {e.printStackTrace();} catch (IOException e) {e.printStackTrace();}}}

批量导入：

Settings settings = Settings.settingsBuilder().put("cluster.name", "es1").build();
TransportClient client = TransportClient.builder().settings(settings).build();
client.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("bigdata2"), 9300));Long count = 100000L;
String index = "bigdata";
String type = "student1";
BulkRequestBuilder bulkRequest = client.prepareBulk();
for (int i = 0; i < count; i++) {Map<String, Object> ret = new HashMap<String, Object>();ret.put("recordtime", 11);ret.put("area", 22);ret.put("usertype", 33);ret.put("count", 44);bulkRequest.add(client.prepareIndex(index, type).setSource(ret));// 每10000条提交一次if (i % 10000 == 0) {bulkRequest.execute().actionGet();}
}

参考文档：

Elasticsearch Reference [2.3] » Document APIs » Bulk API

Elasticsearch java api(五) Bulk批量索引相关推荐

ElasticSearch Java Api(一) －创建索引
ElasticSearch Java API官网文档:https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/jav ...
Elasticsearch RestHighLevelClient 已标记为被弃用它的替代方案 Elasticsearch Java API Client 的基础教程及迁移方案
在Elasticsearch7.15版本之后,Elasticsearch官方将它的高级客户端RestHighLevelClient标记为弃用状态.同时推出了全新的Java API客户端Elastics ...
ElasticSearch Java Api(四) -删除索引
删除可以是删除整个索引库,也可以根据文档id删除索引库下的文档,还可以通过query查询条件删除所有符合条件的数据. 一.删除整个索引库下面的例子会删除indexName索引: DeleteInde ...
Elasticsearch8.0版本中Elasticsearch Java API Client客户端的基本使用方法
关于Elasticsearch Java API Client客户端如何连接以及如何对索引和文档进行基本的增删改查操作请查看我的上一篇博文:Elasticsearch RestHighLevelCli ...
Elasticsearch Java API 很全的整理以及架构剖析
Elasticsearch 的API 分为 REST Client API(http请求形式)以及 transportClient API两种.相比来说transportClient API效率更高, ...
Elasticsearch Java API 6.2（java client）
前言本节描述了Elasticsearch提供的Java API,所有的Elasticsearch操作都使用客户端对象执行,所有操作本质上都是完全异步的(要么接收监听器,要么未来返回). 此外,客户端 ...
Elasticsearch Java API四种实现方式
0.题记之前Elasticsearch的应用比较多,但大多集中在关系型.非关系型数据库与Elasticsearch之间的同步.以上内容完成了Elasticsearch所需要的基础数据量的供给.但想要 ...
SpringBoot整合最新Elasticsearch Java API Client 7.16教程
文章目录前言一.Elasticsearch和Kibana 7.16版本安装二.pom.xml文件引入依赖三.代码实例总结前言最新在学习SpringBoot整合es的一些知识,浏览了网上的 ...
Elasticsearch Java API 分组、聚合、嵌套相关查询
Elasticsearch Java API 分组.聚合.嵌套相关查询翼支付监控系统正使用es做后端存储,这边我们是将日志计算处理过后的数据通过kafka储存到es.选择用es作为数据储存端是考虑到 ...

Elasticsearch java api(五) Bulk批量索引

一、Bulk API

二、批量导出

三、批量导入

Elasticsearch java api(五) Bulk批量索引相关推荐

最新文章

热门文章