ES快速开发，ElasticsearchRestTemplate基本使用以及ELK快速部署 - 文章 - 开发者社区

最近博主有一些elasticsearch的工作，所以更新的慢了些，现在就教大家快速入门，并对一些基本的查询、更新需求做一下示例，废话不多说开始：

1. ES快速上手

es下载：

这里关于es所需要的链接基本都有，可以快速下载使用当你解压好了归档文件之后，Elasticsearch 已经准备好运行了

  
cd elasticsearch-<version>   
 ./bin/elasticsearch

es默认端口9200；transport端口是9300；transport端口主要用来java客户端操作，启动成功后，我们来安装一下kibana界面操作es的工具，也可以下载header，但是kibana界面比较友好并且最后部署ELK的时候也需要该工具，所以博主就直接安装kibana了

kibana下载：

https://elasticsearch.cn/download/

还是跟es一样的链接，下载后解压并编辑config目录下编辑kibana.yml文件，主要配置如下：

  
server.port: 15601  
server.host: "0.0.0.0"  
elasticsearch.hosts: ["http://localhost:9200"]

只需要修改这几处配置就可以，前提是kibana的版本必须与es的版本是相同的，否则会包很多错误，并且启动失败，Linux启动时不能使用root用户启动，必须自己添加个人用户才可以，命令如下：

添加用户：

  
useradd testuser

设置密码：

  
passwd testuser

将我们的文件夹用户权限改变一下要不然启动时候老是提示没有权限：

  
chown -R testuser:testuser kibana

现在进入我们kibana的文件夹，以testuser启动kibana：

  
/bin/kibana

访问地址：http://localhost:15601

当看到这里的时候就已经OK了，我们终于可以开始使用es了。 picture.image

我就不介绍es是干啥用的了，es具有分片的概念，分为主分片和副本分片，创建索引的时候一旦设置副本分片，必须有大于等于2台的机器，每个机器都有es，es之间的交互，需要自己在配置文件中作修改，否则不配置，永远只是单机，并且主分片在建索引的时候必须考虑清楚减多少个主分片，因为以后如果需要修改主分片，必须重新创建索引，你添加或则减少一个主分片，es往分片中存放数据的时候都会变，但是副本分片不一样，因为他是数据冗余的，一旦主分片宕机，副本会当选主分片，并且是要主分片存在，副本没有也可以，副本的作用就是提高数据的吞吐量。好了，开始实战：

点击kibana的Dev Tools按钮，就可以在面板里写语句操作索引了：

建立索引：shards主分片 replicas副本分片设置的数量，看你有几台机器-1。

  
PUT /test  
{  
 "settings": {  
 "number_of_shards": 5,  
 "number_of_replicas": 1  
 },  
 "mappings": {  
 "_doc": {  
 "properties": {  
 "name": {  
 "type": "text",  
 "analyzer": "ik_max_word",  
 "search_analyzer": "ik_smart"  
 },  
 "age": {  
 "type":"integer"  
 }  
 }  
 }  
 }  
}

建立mappings做好字段类型，并且text类型中使用分词器，不要使用默认的分词器，默认的分词器一个汉字分一个，查询出来基本没啥价值，中文分词器是ik，可以上网搜一下下载到es里面。

大家上过语文课，都知道语句有歧义问题，就比如武汉市长江大桥，可以断成武汉市长、江大桥；武汉市、长江大桥；这就是分词器如何切分的问题，所以es有关键字查询term，进行完全匹配，不进行分词器query匹配，除了这些，中文还有同义词一说，比如苹果水果与苹果手机，大家搜索的时候基本都是输入苹果，但是出来的却是苹果手机，水果很少，这就是因为es也可以做同义词查询。但是需要配置同义词文件，具体操作可以自行上网解决，主要就是创建索引的时候，使用自己在config中编辑的文本文件，该文件中有自己要使用到的同义词，比如：iPhone，苹果手机；

我们现在再来进行实战开发，本人接触的是使用ElasticsearchRestTemplate进行开发，该类基本含括了大部分需求开发查询。下面开始举例：

  
//单索引匹配更新   
 Map<String, Object> params = new HashMap<String, Object>();  
 params.put("flag", deleteFlag);  
 //ctx._source即为该索引本身  
 String code = "ctx._source.deleteFlag=params.flag;";  
 ScriptType type = ScriptType.INLINE;  
 //使用脚本进行更新字段值  
 Script script = new Script(type, Script.DEFAULT_SCRIPT_LANG, code, params);  
  
 UpdateByQueryRequest updateByQueryRequest = new UpdateByQueryRequest();  
 updateByQueryRequest.indices("exam_information");//设置索引  
 updateByQueryRequest.setDocTypes("doc");//设置文档，固定doc  
 updateByQueryRequest.setQuery(QueryBuilders.termsQuery("paperBaseId", paperBaseId));//设置查询  
 updateByQueryRequest.setScript(script);//如果有脚本，则添加  
 updateByQueryRequest.setConflicts("proceed"); // 设置版本冲突时继续  
 updateByQueryRequest.setRefresh(true);//请求结束后，对我们写入的索引进行调用刷新  
 this.elasticsearchTemplate.getClient().updateByQuery(updateByQueryRequest, RequestOptions.DEFAULT);//进行更新  
   
 //多索引匹配批量更新  
 Map<String,Object> updateMap = new HashMap<>();  
 updateMap.put("deleteFlag",deleteFlag);  
 updateMap.put("lastUpdateTime",currDatetime);  
 UpdateRequest doc = new UpdateRequest().doc(updateMap);  
 List<UpdateQuery> updateQuerys = new ArrayList<>();  
 //生成批量更新操作  
 paperBaseId.stream().forEach(id ->{  
 UpdateQuery build = new UpdateQueryBuilder()  
 .withUpdateRequest(doc)  
 .withDoUpsert(true)  
 .withIndexName("paper_base")  
 .withType("doc")  
 .withId(id).build();  
 updateQuerys.add(build);  
 });  
 elasticsearchTemplate.bulkUpdate(updateQuerys,BulkOptions.builder().withRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).build());  
   
 //查询操作  
 MatchQueryBuilder lastUpdateUser = QueryBuilders.matchQuery("personId", userId);  
 MatchQueryBuilder deleteflag = QueryBuilders.matchQuery("deleteFlag", BaseEntity.DEL_FLAG_DELETE);  
 //创建bool多条件查询  
 BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();  
 BoolQueryBuilder mustQuery = boolQueryBuilder.must(lastUpdateUser).must(deleteflag);  
 //嵌套索引，需要使用nest查询  
 mustQuery.must(QueryBuilders.nestedQuery("entityNodes", QueryBuilders.termQuery("entityNodes.node_type", recyclePaperDTO.getNodeType()), ScoreMode.None));  
 //可以使用should查询，不是必需条件  
 BoolQueryBuilder nodeQueryBuilder = QueryBuilders.boolQuery();  
 nodeQueryBuilder.should(QueryBuilders.nestedQuery("entityNodes", QueryBuilders.wildcardQuery("entityNodes.parent_ids", "*," + recyclePaperDTO.getNodeId() + "*"), ScoreMode.None));  
 nodeQueryBuilder.should(......);  
 mustQuery.must(nodeQueryBuilder);  
 //查询使用排序  
 SortBuilder order = new FieldSortBuilder("lastUpdateTime").order(SortOrder.DESC);  
 //可以使用高亮显示，就是html标签  
 HighlightBuilder highlightBuilder = new HighlightBuilder();  
 highlightBuilder.preTags("<span class='highlighted'>")  
 .postTags(</span>)  
 .field("paperBaseName");//哪个字段高亮  
 //使用分页查询  
 SearchQuery nativeSearchQueryBuilder = new NativeSearchQueryBuilder()  
 .withQuery(mustQuery).withSort(order).withHighlightBuilder(highlightBuilder)  
 .withPageable(PageRequest.of(recyclePaperDTO.getPageNum()-1, recyclePaperDTO.getPageSize())).build();  
 //进行查询，entityMapper使用默认的也可，EsPaperBase.class是需要自己映射的查询类  
 elasticsearchTemplate.queryForPage(nativeSearchQueryBuilder, EsPaperBase.class, new HighlightResultMapper(entityMapper));

  
@Data  
@Builder  
@NoArgsConstructor  
@AllArgsConstructor  
@Document(indexName = "paper_base", type = "doc")  
@Setting(settingPath = "/elasticsearch/settings.json")//可设置主分片、副本分片、设置默认停用词等  
public class EsPaperBase {  
  
 @Id  
 @Field(type = FieldType.Keyword, name = "paperBaseId")  
 private String paperBaseId;  
  
 /**  
 * 试卷名称  
 */  
 @MultiField(mainField = @Field(type = FieldType.Text, analyzer = "standard" , name = "paperBaseName"),  
 otherFields = {  
 @InnerField(suffix = "zh", type = FieldType.Text, analyzer = "ik_max_word", searchAnalyzer = "ik_smart"),  
 @InnerField(suffix = "en", type = FieldType.Text, analyzer = "english"),  
 })  
 private String paperBaseName;  
  
 /**  
 * 共享级别名,可以使用分词器查询，模糊匹配  
 */  
 @Field(type = FieldType.Text, name = "shareLevelName")  
 private String shareLevelName;  
  
  
 /**  
 * 创建人，不可使用分词器查询，精准匹配  
 */  
 @Field(type = FieldType.Keyword, name = "personId")  
 private String personId;  
  
  
 /**  
 * 创建时间  
 */  
 @Field(type = FieldType.Date, name = "createtime", format = DateFormat.custom, pattern = "yyyy-MM-dd HH:mm:ss")  
 private String createtime;  
  
 /**  
 * 更新时间  
 */  
 @Field(type = FieldType.Date, name = "lastUpdateTime", format = DateFormat.custom, pattern = "yyyy-MM-dd HH:mm:ss")  
 private String lastUpdateTime;  
  
 /**  
 * 删除标识 0:未删除，1:已删除  
 */  
 @Field(type = FieldType.Keyword, name = "deleteFlag")  
 private String deleteFlag;  
 /**  
 * 试卷推荐，内嵌字段  
 */  
 @Field(type=FieldType.Nested,name="paperRecommends")  
 private List<EsPaperRecommend> paperRecommends;  
   
 ......  
}

  
setting.json  
{  
 "index": {  
 "number_of_shards": "5",  
 "number_of_replicas": "2",  
 "refresh_interval": "1s",  
 "max_rescore_window": 10000000  
 },  
 "analysis": {  
 "filter": {  
 "spanish_stop": {  
 "type": "stop",  
 "stopwords": [ "si", "esta", "el", "la" ]   
 },  
 "light_spanish": {   
 "type": "stemmer",  
 "language": "light_spanish"  
 }  
 },  
 "analyzer": {  
 "my_spanish": {  
 "tokenizer": "spanish",  
 "filter": [ //顺序很重要  
 "lowercase",  
 "asciifolding",  
 "spanish_stop",  
 "light_spanish"  
 ]  
 }  
 }  
 }  
 }

现在很多公司基本使用分布式架构应用，公司每个应用模块都有好几台机器，看日志问题也就衍生而来，我们最笨的方法就是每个服务器后台都打开进行查看，效率低下，此时，我们就可以使用es、kibana、logstash；简称ELK进行查看分布式日志系统，但是本文不会进行安装logstash进行演示，因为只做日志查询的需求，我们使用ELK的变种EFK即可，filebeat轻量级做日志收集即可，最主要的就是看我们如何进行配置，然后使用kibana进行查询日志。

安装完logstash后，解压在config中新建my-logstash.conf，该配置中注意三大块，input、filter、output；其中input是作为吸取日志的以.log为后缀的日志文件，filter是过滤配置，不用管，output则是导入到哪个elasticsearch中；配置如下：

  
input {  
 file {  
 type => "log"  
 path => ["/apps/svr/server/*/log.file"]  
 start_position => "end"  
 ignore_older => 0  
 codec=> multiline {  
 pattern => "^\d{4}-\d{1,2}-\d{1,2}\s\d{1,2}:\d{1,2}:\d{1,2}"  
 negate => true  
 auto_flush_interval => 5  
 what => "previous"  
 }  
 }  
 beats {  
 port => 5044  
 }  
}  
output {  
 if [type] == "log" {  
 elasticsearch {  
 hosts => ["http://127.0.0.1:19200"]  
 index => "logstash-%{+YYYY.MM}"  
 #user => es  
 #password => es2018  
 }   
 }  
}

如果自己动手配置的话，最好自己手动输入，不要复制粘贴，很有可能会有特殊字符出现导致启动失败；启动命令：./bin/logstah -f my-logstash.conf picture.image

最终我们就可以这样使用kibana进行查询日志的操作了。简单的基本应用就到此为止了，工作中基础的应用是没有问题了；最后记得关注本博主的公众号啊！