实时搜索引擎Elasticsearch(5)——Java API的使用

前一篇有关ES的文章介绍了使用Rest方式调用ES的聚合API。Rest API使用了HTTP协议,按理来说,可以直接使用类似HttpClient的工具直接调用Rest API。虽然笔者并没有尝试过,但稍微想想一下就知道这种方法是可行的。这种方法主要有下面几个弊端:

  1. 需要开启ES的Http服务和端口。ES提供的Http服务功能非常全面,没有提供权限控制,防护也比较脆弱。一旦遭到破解,则数据面临极大的风险。所以,建议在生产中关闭Http服务,或者自己增加一层代理来实现权限控制。
  2. 调用比较困难。Rest API的核心是url和post数据,url直接需传入字符串,这样就不能使用IDE的查错功能。需要记忆的东西太多,不确定时就要去查API,影响开发效率。
  3. Http协议的一大特点是无连接性。也就是每一次请求都需要建立新的连接,我们知道tcp连接是比较耗时的过程。从性能的角度来说,直接使用Rest API也是不合适的。

ES所提供的Http服务适合用作集群状态和数据的监控,而不适合直接用于数据操作。ES提供了多种语言(包括Java、Python、PHP、Ruby等)版本的Client API,可以使用这些Client API编程实现数据操作功能。作为一个Java语言编程者,本文主要介绍使用Java版本的Client来操作数据。

使用Java API需要依赖ES所提供的jar包,我们使用maven来下载所需的依赖包,maven依赖定义如下:

1
2
3
4
5
<dependency>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch</artifactId>
<version>1.5.0</version>
</dependency>

version表示依赖包的版本,可以输入任意存在的版本,本文的示例中使用1.5.0版的API。注意,建议API的版本与ES集群所使用的版本保持一致,以免出现因版本不一致而导致的冲突。

本文的主要内容包括:

  1. 介绍两类Client,解释它们的之间的差异;
  2. 使用Client进行index、document和聚合相关的操作。

Client

ES中所有的Java API调用都要使用Client对象,ES为API调用者提供了两类Client对象:NodeClient和TransportClient。下面来讲讲这两类Client的差异和使用场景。

NodeClient

NodeClient是一种嵌入式节点客户端。它首先在客户端启动一个节点(Node),并加入同名集群内。这个节点可以保存数据,并且数据能够被索引。然后从这个节点中获取Client,这类Client就是NodeClient。NodeClient无需指明ES服务端的地址,操作的数据位于启动的节点所在的集群中。下面是获得NodeClient的代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import org.elasticsearch.client.Client;
import org.elasticsearch.client.node.NodeClient;
import org.elasticsearch.node.Node;
import static org.elasticsearch.node.NodeBuilder.nodeBuilder;
public class MyNodeClient {
public static void main(String[] args) {
// 启动一个本地节点,并加入子网内的ES集群
Node node = nodeBuilder()
.clusterName("elasticsearch") // 要加入的集群名为elasticsearch
// .client(true) //如果设置为true,则该节点不会保存数据
.data(true) // 本嵌入式节点可以保存数据
.node(); // 构建并启动本节点
// 获得一个Client对象,该对象可以对子网内的“elasticsearch”集群进行相关操作。
Client nodeClient = node.client();
}
}

运行这段代码之后,可以看到工程中新增了一个data文件夹,这是因为data(true)将Node设置为可以存放数据的节点,数据正是放在了data文件夹下。

新增加的data文件夹

NodeClient适合用作单元或集成测试,而不适合用于生产环境。

TransportClient

TransportClient连接远端的ES集群,其本身并不会加入集群。创建TransportClient的代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
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.common.transport.TransportAddress;
import static org.elasticsearch.common.settings.ImmutableSettings.settingsBuilder;
public class MyTransportClient {
public static void main(String[] args) {
// 配置信息
Settings esSetting = settingsBuilder()
.put("cluster.name", "elasticsearch")
.build();
TransportClient transportClient = new TransportClient(esSetting);
// 添加连接地址
TransportAddress address = new InetSocketTransportAddress("192.168.1.110", 9300);
TransportAddress address2 = new InetSocketTransportAddress("192.168.1.111", 9300);
transportClient.addTransportAddress(address);
transportClient.addTransportAddress(address2);
}
}

TransportClient适合用于生产环境中。

Index操作

本小节介绍如果使用Java API创建和删除索引。

创建索引

废话先不说,上代码先。下面的方法创建一个索引,并同时创建一个mapping。mapping可以传入符合格式要求的json字符串。一般情况下,我们可以使用下面的方式来生成所需的json字符串。

  1. 手动拼接json字符串
  2. 使用类似jackson的工具将对象转换为相应的json字符串
  3. 使用ES内置的XContentFactory.jsonBuilder()来创建json字符串。

本文的示例中均使用ES自带的XContentFactory.jsonBuilder()来构建json字符串。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
/**
* 创建一个索引
* @param indexName 索引名
*/
public void createIndex(String indexName) {
try {
CreateIndexResponse indexResponse = this.client
.admin()
.indices()
.prepareCreate(indexName)
.get();
System.out.println(indexResponse.isAcknowledged()); // true表示创建成功
} catch (ElasticsearchException e) {
e.printStackTrace();
}
}

如果需要再索引上新建mapping,可通过下面的代码来实现。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
/**
* 给索引增加mapping。
* @param index 索引名
* @param type mapping所对应的type
*/
public void addMapping(String index, String type) {
try {
// 使用XContentBuilder创建Mapping
XContentBuilder builder =
XContentFactory.jsonBuilder()
.startObject()
.field("properties")
.startObject()
.field("name")
.startObject()
.field("index", "not_analyzed")
.field("type", "string")
.endObject()
.field("age")
.startObject()
.field("index", "not_analyzed")
.field("type", "integer")
.endObject()
.endObject()
.endObject();
System.out.println(builder.string());
PutMappingRequest mappingRequest = Requests.putMappingRequest(index).source(builder).type(type);
this.client.admin().indices().putMapping(mappingRequest).actionGet();
} catch (ElasticsearchException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
}

删除索引

1
2
3
4
5
6
7
8
9
10
11
12
13
/**
* 删除索引
* @param index 要删除的索引名
*/
public void deleteIndex(String index) {
DeleteIndexResponse deleteIndexResponse =
this.client
.admin()
.indices()
.prepareDelete(index)
.get();
System.out.println(deleteIndexResponse.isAcknowledged()); // true表示成功
}

文档CURD操作

增删改查是数据的基本操作,同时也是使用频率最高的一类操作。本小节介绍使用Java API来实现document的增删改查。

新增文档

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
/**
* 创建一个文档
* @param index index
* @param type type
*/
public void createDoc(String index, String type) {
try {
// 使用XContentBuilder创建一个doc source
XContentBuilder builder =
XContentFactory.jsonBuilder()
.startObject()
.field("name", "zhangsan")
.field("age", "lisi")
.endObject();
IndexResponse indexResponse = this.client
.prepareIndex()
.setIndex(index)
.setType(type)
// .setId(id) // 如果没有设置id,则ES会自动生成一个id
.setSource(builder.string())
.get();
System.out.println(indexResponse.isCreated());
} catch (ElasticsearchException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
}

更新文档

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
/**
* 更新文档
* @param index
* @param type
* @param id
*/
public void updateDoc(String index, String type, String id) {
try {
XContentBuilder builder =
XContentFactory.jsonBuilder()
.startObject()
.field("name", "lisi")
.field("age", 12)
.endObject();
UpdateResponse updateResponse =
this.client
.prepareUpdate()
.setIndex(index)
.setType(type)
.setId(id)
.setDoc(builder.string())
.get();
System.out.println(updateResponse.isCreated()); // true表示成功
} catch (ElasticsearchException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
}

注意,id参数必须是确定存在的id值,否则会抛出document missing的异常。

查询文档

查询文档可以是根据id查询,也可以是根据复杂的查询条件查询。根据id的get查询代码如下。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
/**
* 根据ID查询一条数据记录。
* @param id 要查询数据的ID。
* @return 返回查询出来的记录对象的json字符串。
*/
public String get(String index, String type, String id) {
GetResponse getResponse = this.client
.prepareGet() // 准备进行get操作,此时还有真正地执行get操作。(与直接get的区别)
.setIndex(index) // 要查询的
.setType(type)
.setId(id)
.get();
return getResponse.getSourceAsString();
}
`

基于复杂查询条件的示例代码如下。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
/**
* 使用filter方式查询数据。
* @param index 数据所在的索引名
* @param type 数据所在的type
* @return
*/
public List<String> queryByFilter(String index, String type) {
// 查询名为zhangsan的数据
FilterBuilder filterBuilder = FilterBuilders.termFilter("name", "zhangsan");
SearchResponse searchResponse =
this.client
.prepareSearch()
.setIndices(index)
.setTypes(type)
.setPostFilter(filterBuilder)
.get();
List<String> docList = new ArrayList<String>();
SearchHits searchHits = searchResponse.getHits();
for (SearchHit hit : searchHits) {
docList.add(hit.getSourceAsString());
}
return docList;
}

删除文档

下面的代码删除指定id的文档。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
/**
* 删除一条数据
* @param index
* @param type
* @param id
*/
public void deleteDoc(String index, String type, String id) {
DeleteResponse deleteResponse = this.client
.prepareDelete()
.setIndex(index)
.setType(type)
.setId(id)
.get();
System.out.println(deleteResponse.isFound()); // true表示成功
}

根据复杂的查询条件来删除文档。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
/**
* 根据查询条件删除文档。
*/
public void deleteByQuery(String index, String type) {
try {
QueryBuilder queryBuilder = QueryBuilders.termQuery("name", "zhangsan");
DeleteByQueryResponse deleteByQueryResponse = this.client
.prepareDeleteByQuery(index)
.setTypes(type)
.setQuery(queryBuilder)
.get();
} catch (ElasticsearchException e) {
e.printStackTrace();
}
}

4. 聚合操作

聚合操作的API稍微比较复杂一点,本文仅以min聚合的示例来说明聚合API的调用方式,其他的聚合API调用步骤类似。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
/**
* 使用min聚合查询某个字段上最小的值。
* @param index
* @param type
*/
public void min(String index, String type) {
SearchResponse response = this.client
.prepareSearch(index)
.addAggregation(AggregationBuilders.min("min").field("age"))
.get();
InternalMin min = response.getAggregations().get("min");
System.out.println(min.getValue());
}


源代码地址:https://github.com/xialei199023/blog-demo