ELK使用记录-Elasticsearch使用记录

￡神魔★判官ぃ 2022-11-03 11:49 159阅读 0赞

> Java RestHighLevelClient 操作

# 分组聚合带条件 #

//查询es 书籍日志
            SearchRequest searchRequest = new SearchRequest("bookanalysis");
            BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
            List<QueryBuilder> filter = boolQuery.filter();
            long start = LocalDateTime.now().plusDays(-1).withHour(0).withMinute(0).withSecond(0).withNano(0).toInstant(ZoneOffset.of("+8")).toEpochMilli();
    //        long start = LocalDate.now().plusDays(-1).atStartOfDay(ZoneId.systemDefault()).toInstant().toEpochMilli();
            long end = LocalDate.now().atStartOfDay(ZoneId.systemDefault()).toInstant().toEpochMilli();
    
            //开始时间
            Long startMilli = start;
            //结束时间
            Long endMilli = end;
            //时间区间条件
            filter.add(rangeQuery(startMilli, endMilli, "create_time"));
            //表示是阅读书籍数据
            filter.add(QueryBuilders.matchQuery("log_type",0));
    
            //这里拼接动态条件
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
    
            //添加条件
            searchSourceBuilder.query(boolQuery);
            //时间分桶
            DateHistogramAggregationBuilder builder = AggregationBuilders
                    .dateHistogram("create_time")
                    .field("create_time")
                    .calendarInterval(DateHistogramInterval.HOUR).format("yyyy-MM-dd HH:mm:ss");
    
            //书籍桶
            TermsAggregationBuilder book_idField = AggregationBuilders.terms("book_id").field("book_id");
            TermsAggregationBuilder chapter_idField = AggregationBuilders.terms("content_id").field("content_id");
            //将子桶加入桶中
            builder.subAggregation(book_idField.subAggregation(chapter_idField));
    
            //阅读章节数
    //          book_idField.subAggregation(AggregationBuilders.count("content_id_count").field("content_id"));
            //阅读UV
            chapter_idField.subAggregation(AggregationBuilders.cardinality("read_uv").field("user_id"));
    
            searchSourceBuilder.aggregation(builder);
            searchRequest.source(searchSourceBuilder);
            SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
    
    
            //每日的
            List<AppStatisticsChapterUvEveryday> everydayList = new ArrayList<>();
            //累计的
            List<AppStatisticsChapterUvTotal> totalList = new ArrayList<>();
    
    
            ParsedDateHistogram terms = response.getAggregations().get("create_time");
            Iterator<? extends Histogram.Bucket> timeIter = terms.getBuckets().iterator();
            while (timeIter.hasNext()){
                Histogram.Bucket timeBucket = timeIter.next();
                Object create_time = timeBucket.getKey();
                ParsedLongTerms bookTerms = timeBucket.getAggregations().get("book_id");
                Iterator<? extends Terms.Bucket> bookIter = bookTerms.getBuckets().iterator();
                while (bookIter.hasNext()){
                    Terms.Bucket bookBucket = bookIter.next();
                    int book_id = bookBucket.getKeyAsNumber().intValue();
                    ParsedLongTerms contentTerms = bookBucket.getAggregations().get("content_id");
                    Iterator<? extends Terms.Bucket> contentIter = contentTerms.getBuckets().iterator();
                    while (contentIter.hasNext()) {
                        Terms.Bucket contentBucket = contentIter.next();
                        int chapter_id = contentBucket.getKeyAsNumber().intValue();
                        Aggregations aggregations = contentBucket.getAggregations();
                        ZonedDateTime date = (ZonedDateTime) create_time;
    
                        AppStatisticsChapterUvEveryday chapterUvEveryday = new AppStatisticsChapterUvEveryday();
                        Date createTime = new SimpleDateFormat("yyyy-MM-dd").parse(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
                                .format(date.toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime()));
                        chapterUvEveryday.setStatistics_date(createTime);
                        //阅读UV
                        ParsedCardinality read_uv = aggregations.get("read_uv");
                        chapterUvEveryday.setRead_uv((int) read_uv.getValue());
                        //书籍id
                        chapterUvEveryday.setBook_id(book_id);
                        //章节id
                        chapterUvEveryday.setChapter_id(chapter_id);
                        //章节名称
                        AppBookChapter appBookChapter = appBookChapterService.chapterInfoById(chapter_id);
                        if (appBookChapter==null){
                            continue;
                        }
                        chapterUvEveryday.setChapter_name(appBookChapter.getTitle());
                        everydayList.add(chapterUvEveryday);
                        //查询上次的累计内容
                        AppStatisticsChapterUvTotal one = appStatisticsChapterUvTotalService.getOne(
                                new QueryWrapper<AppStatisticsChapterUvTotal>()
                                        .eq("chapter_id", chapter_id)
                                        .orderByDesc("statistics_date")
                                        .last("limit 0,1")
                        );
                        if (one == null) {
                            one = new AppStatisticsChapterUvTotal();
                            one.setStatistics_date(createTime);
                            //阅读UV
                            one.setRead_uv((int) read_uv.getValue());
                            //书籍id
                            one.setBook_id(book_id);
                            //章节id
                            one.setChapter_id(chapter_id);
                            //章节名称
                            one.setChapter_name(appBookChapter.getTitle());
                        }else{
                            one.setId(null);
                            one.setStatistics_date(createTime);
                            //阅读UV
                            one.setRead_uv(one.getRead_uv()+ (int) read_uv.getValue());
                            //书籍id
                            one.setBook_id(book_id);
                            //章节id
                            one.setChapter_id(chapter_id);
                            //章节名称
                            one.setChapter_name(appBookChapter.getTitle());
                        }
                        totalList.add(one);
                    }
                }
            }
            appStatisticsChapterUvTotalService.saveBatch(totalList);
            appStatisticsChapterUvEverydayService.saveBatch(everydayList);

# 获取桶内的真正数据hits #

TopHitsAggregationBuilder topHitsAggregationBuilder = AggregationBuilders.topHits("top");
    build.subAggregation(topHitsAggregationBuilder);
    
    
    TopHits topHits = Bucket.getAggregations().get("top");
    for (SearchHit hit : topHits.getHits().getHits()) {
        hit.getSourceAsMap().get("device_id");
    }
    
    
    //获取超过1w条数据 需要加上  "track_total_hits":true ，不然只能显示出9999条
    searchRequest.source(searchSourceBuilder.trackTotalHits(true)); 
    
    
    
    //博客学习
    //内容: Top Hits Aggregation
    //网址: https://blog.csdn.net/ctwy291314/article/details/82773180

# 普通查询,使用查询内容 #

@Test
        public void test() throws IOException {
            SearchRequest searchRequest = new SearchRequest("useranalysis");
            BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
            List<QueryBuilder> filter = boolQuery.filter();
            //时间区间条件
            filter.add(rangeQuery(1611590400000L, 1611676800000L, "create_time"));
            filter.add(QueryBuilders.matchQuery("channel_id",50000011104L));
            filter.add(QueryBuilders.matchQuery("is_new",1));
            //这里拼接动态条件
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            //添加条件
            searchSourceBuilder.query(boolQuery);
    
            searchRequest.source(searchSourceBuilder.trackTotalHits(true).size(20001));
            SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            HashSet<String> userIds = new HashSet<>();
            for (SearchHit hit : response.getHits().getHits()) {
                String user_id = hit.getSourceAsMap().get("user_id").toString();
                userIds.add(user_id);
            }
            System.out.println(userIds);
        }

# 使用模板查询(dsl语言) #

public int test(List<Integer> userIds) throws IOException {
            long start = LocalDateTime.now().plusDays(-1).withHour(0).withMinute(0).withSecond(0).withNano(0).toInstant(ZoneOffset.of("+8")).toEpochMilli();
            long end = LocalDateTime.now().plusDays(0).withHour(0).withMinute(0).withSecond(0).withNano(0).toInstant(ZoneOffset.of("+8")).toEpochMilli();
            SearchTemplateRequest request = new SearchTemplateRequest();
            request.setRequest(new SearchRequest("bookanalysis"));
            request.setScriptType(ScriptType.INLINE);
            String script = "{\n" +
                    "  \"query\": {\n" +
                    "    \"bool\": {\n" +
                    "      \"must\": [\n" +
                    "        {\n" +
                    "          \"range\": {\n" +
                    "            \"create_time\": {\n" +
                    "              \"gte\": \"{
        {start}}\",\n" +
                    "              \"lte\": \"{
        {end}}\"\n" +
                    "            }\n" +
                    "          }\n" +
                    "        },\n" +
                    "        {\n" +
                    "          \"terms\": {\n" +
                    "            \"user_id\":{
        {user_id}}\n" +
                    "          }\n" +
                    "        }\n" +
                    "      ]\n" +
                    "    }\n" +
                    "  },\n" +
                    "  \"size\": 0,\n" +
                    "  \"track_total_hits\": true"+
                    "}";
            request.setScript(script);
            Map<String, Object> params = new HashMap<>();
            params.put("user_id", JSONObject.toJSONString(userIds));
            params.put("start", start);
            params.put("end", end);
            request.setScriptParams(params);
            SearchTemplateResponse searchTemplateResponse = restHighLevelClient.searchTemplate(request, RequestOptions.DEFAULT);
            SearchResponse response = searchTemplateResponse.getResponse();
            TotalHits totalHits = response.getHits().getTotalHits();
            return (int) totalHits.value;
        }

# 删除数据 #

public ReturnT<String> deleteEveryday(String param) throws IOException {
            //删除指定的传入的索引
            try {
                DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest(param);
                BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
                List<QueryBuilder> filter = boolQueryBuilder.filter();
                long start = LocalDateTime.now().plusDays(-10).withHour(23).withMinute(59).withSecond(59).withNano(0).toInstant(ZoneOffset.of("+8")).toEpochMilli();
                long end = LocalDateTime.now().plusDays(-2).withHour(0).withMinute(0).withSecond(0).withNano(0).toInstant(ZoneOffset.of("+8")).toEpochMilli();
                XxlJobLogger.log("\t 开始删除数据:("+start+"-"+end+") 索引地址:"+param+"。\t 开始时间:"+LocalDateTime.now().toString());
                filter.add(rangeQuery(start, end, "create_time"));
                deleteByQueryRequest.setQuery(boolQueryBuilder);
                //并行 5条线程同时执行
                deleteByQueryRequest.setSlices(5);
                //批次大小 一次删除多少
                deleteByQueryRequest.setBatchSize(5000);
                // 更新最大文档数
    //            deleteByQueryRequest.setSize(10);
                //使用滚动参数来控制“搜索上下文”存活的时间  因为删除数据前是先查询到这个数据
                deleteByQueryRequest.setScroll(TimeValue.timeValueMinutes(10));
                //刷新索引
                deleteByQueryRequest.setRefresh(true);
                //超时时间 2小时
                deleteByQueryRequest.setTimeout(TimeValue.timeValueHours(2));
                BulkByScrollResponse response = restHighLevelClient.deleteByQuery(deleteByQueryRequest,RequestOptions.DEFAULT);
                XxlJobLogger.log("\t 删除数据:"+response.getTotal()+" 条"+"\t 结束时间:"+LocalDateTime.now().toString());
                            //使用异步删除
                /*
                Cancellable response = restHighLevelClient.deleteByQueryAsync(
                        deleteByQueryRequest,
                        RequestOptions.DEFAULT,
                        new ActionListener<BulkByScrollResponse>() {
                            @Override public void onResponse(BulkByScrollResponse response) {
    //                            XxlJobLogger.log("\t 删除数据:"+response.getTotal()+" 条"+"\t 异步结束时间:"+LocalDateTime.now().toString());
                                System.out.println("\t 删除数据:"+response.getTotal()+" 条"+"\t 异步结束时间:"+LocalDateTime.now().toString());
                            }
                            @Override public void onFailure(Exception e) {
                                e.printStackTrace();
    //                            XxlJobLogger.log("\t 异步删除数据报错了........"+LocalDateTime.now().toString()+"\t"+e.getMessage());
                                System.out.println("\t 异步删除数据报错了........"+LocalDateTime.now().toString()+"\t"+e.getMessage());
                            }
                        });
                //取消删除
    //            response.cancel();
    //            Thread.sleep(2000);
                 */
            } catch (Exception e) {
                e.printStackTrace();
                XxlJobLogger.log("\t 删除数据报错了........"+LocalDateTime.now().toString()+"\t"+e.getMessage());
                return ReturnT.FAIL;
            }
            return ReturnT.SUCCESS;