Java elasticsearch 基本使用

今天使用到，所以记录了下来，可以搭配 kinaba 使用会更加方便，废话少说。
以下示例是一个解析日志的场景

1. 添加依赖

<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-high-level-client</artifactId>
    <version>7.3.2</version>
</dependency>

2. 初始化客户端

    private static final RestHighLevelClient restHighLevelClient = new RestHighLevelClient(
            RestClient.builder(
                    new HttpHost("127.0.0.1", 9200, "http")
            )
    );

不是长连接的话，最好可以有 close 函数

    public void close(RestHighLevelClient client) {
        LOGGER.info("Closing elasticSearch client");
        if (client != null) {
            try {
                client.close();
            } catch (IOException e) {
                LOGGER.error("Error closing ElasticSearch client: " + e.getMessage());
            }
        }
    }

3. 查询某条记录的详情

kinaba DTS

GET http-log/kong/hp8rpXABLfkHYzAW6xxf

java

        if (index != null && type != null && id != null) {
   
            try {
                GetRequest getRequest = new GetRequest(index, type, id);
                GetResponse getResponse = restHighLevelClient.get(getRequest);
                if (getResponse.isExists()) {
                    List<Map<String, Object>> list = new ArrayList<Map<String, Object>>();
                    Map<String, Object> map = getResponse.getSourceAsMap();
                    for (Map.Entry<String, Object> entry : map.entrySet()) {
                        if (entry.getKey() == "started_at") {
                            map.put("started_at", timeStampToDate(entry.getValue()));
                        }
                    }
                    list.add(map);
                    return list;
                }
            } catch (IOException e) {
                LOGGER.info("catch IOException error: " + e.getMessage());
            }
        }

4. 多条件查询: 获取某个时间段的 get 请求，不返回 hits 数据

kinaba DTS

GET /http-log/kong/_search
{
  "size": 0,
  "query": {
    "bool": {
      "must": [
        {
          "match": {
            "request.uri": "/get"
          }
        },
        {
          "range": {
            "started_at": {
              "gte": 1573429400000,
              "lte": 1583852400000
            }
          }  
        }
      ]
    }
  } 
}

java

        // 开始时间 -> 结束时间 + request uri 查询
        if (index != null && type != null && startTime != null && endTime != null && path != null) {

            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();

            searchSourceBuilder = paging(searchSourceBuilder, number, page);
            searchSourceBuilder.timeout(new TimeValue(60, TimeUnit.SECONDS));


            RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery("started_at").gte(startTime).lte(endTime);
            searchSourceBuilder.sort(new FieldSortBuilder("started_at").order(SortOrder.DESC));
            TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery(ConstSet.REQUEST_URI, path);

            BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery().must(rangeQueryBuilder).must(termQueryBuilder);

            searchSourceBuilder.query(boolQueryBuilder);

            SearchRequest searchRequest = new SearchRequest(index);
            searchRequest.types(type);
            searchRequest.source(searchSourceBuilder);
            try {
                SearchResponse response = restHighLevelClient.search(searchRequest);
                SearchHits searchHits = response.getHits();

                long totalHits = searchHits.getTotalHits();
                LOGGER.info("totalHits value is: " + totalHits);

                List<Map<String, Object>> list = new ArrayList<Map<String, Object>>();

                for (SearchHit hit : searchHits) {
                    Map<String, Object> map = hit.getSourceAsMap();
                    map.put("total", totalHits);
                    for (Map.Entry<String, Object> entry : map.entrySet()) {
                        if (entry.getKey() == "started_at") {
                            map.put("started_at", timeStampToDate(entry.getValue()));
                        }
                    }
                    list.add(hit.getSourceAsMap());
                }
                return list;
            } catch (IOException e) {
                LOGGER.info("catch IOException error: " + e.getMessage());
            }
        }

5. 组合聚合查询，按照时间范围对请求方式进行分组

kinaba DTS

GET /http-log/kong/_search
{
  "query": {
    "range": {
      "started_at": {
        "gte": 1553515800000,
        "lte": 1593852400000
      }
    }
  },
  "size": 0, 
  "aggs": {
    "group": {
      "terms": {
        "field": "request.method.keyword",
        "size": 10
      }
    }
  }
}

java

      if (index != null && type != null && startTime != null && endTime != null && path != null) {
      
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();

            searchSourceBuilder = paging(searchSourceBuilder, number, page);
            searchSourceBuilder.timeout(new TimeValue(60, TimeUnit.SECONDS));


            RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery("started_at").gte(startTime).lte(endTime);
            searchSourceBuilder.sort(new FieldSortBuilder("started_at").order(SortOrder.DESC));
            TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery(ConstSet.REQUEST_URI, path);

            BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery().must(rangeQueryBuilder).must(termQueryBuilder);

            searchSourceBuilder.query(boolQueryBuilder);

            SearchRequest searchRequest = new SearchRequest(index);
            searchRequest.types(type);
            searchRequest.source(searchSourceBuilder);
            try {
                SearchResponse response = restHighLevelClient.search(searchRequest);
                SearchHits searchHits = response.getHits();

                long totalHits = searchHits.getTotalHits();
                LOGGER.info("totalHits value is: " + totalHits);

                List<Map<String, Object>> list = new ArrayList<Map<String, Object>>();

                for (SearchHit hit : searchHits) {
                    Map<String, Object> map = hit.getSourceAsMap();
                    map.put("total", totalHits);
                    for (Map.Entry<String, Object> entry : map.entrySet()) {
                        if (entry.getKey() == "started_at") {
                            map.put("started_at", timeStampToDate(entry.getValue()));
                        }
                    }
                    list.add(hit.getSourceAsMap());
                }
                return list;
            } catch (IOException e) {
                LOGGER.info("catch IOException error: " + e.getMessage());
            }
        }