ELK（使用踩坑）

忘是亡心i 2023-01-10 01:21 77阅读 0赞

E:elastic  
L:logstash  
K:kibnan

es配置

elastic.host=服务器地址
    elastic.port=9200
    elastic.scheme=http
    elastic.username=名称
    elastic.password=密码

es工具类：

@Component
    @ConditionalOnBean(value = RestHighLevelClient.class)
    public class ElasticUtil {
    
        @Autowired
        RestHighLevelClient restHighLevelClient;
    
        /**
         * @param idxName   索引名称
         * @param idxSQL    索引描述
         * @return void
         * @throws Exception 
         * @throws
         * @since
         */
        public void createIndex(String idxName,String idxSQL) throws Exception{
            if (!this.indexExist(idxName)) {
                return;
            }
            CreateIndexRequest request = new CreateIndexRequest(idxName);
            buildSetting(request);
            request.mapping(idxSQL, XContentType.JSON);
            CreateIndexResponse res = restHighLevelClient.indices().create(request, RequestOptions.DEFAULT);
            if (!res.isAcknowledged()) {
                throw new RuntimeException("初始化失败");
            }
        }
    
        /** 断某个index是否存在
         * @param idxName index名
         * @return boolean
         * @throws
         * @since
         */
        public boolean indexExist(String idxName) throws Exception {
            GetIndexRequest request = new GetIndexRequest(idxName);
            request.local(false);
            request.humanReadable(true);
            request.includeDefaults(false);
            request.indicesOptions(IndicesOptions.lenientExpandOpen());
            return restHighLevelClient.indices().exists(request, RequestOptions.DEFAULT);
        }
    
        /** 断某个index是否存在
         * @throws
         * @since
         */
        public boolean isExistsIndex(String idxName) throws Exception {
            return restHighLevelClient.indices().exists(new GetIndexRequest(idxName),RequestOptions.DEFAULT);
        }
    
        /** 设置分片
         * @since
         */
        public void buildSetting(CreateIndexRequest request){
            request.settings(Settings.builder().put("index.number_of_shards",3)
                    .put("index.number_of_replicas",2));
        }
        /**
         * @throws IOException 
         * @throws
         * @since
         */
        public void insertOrUpdateOne(String idxName, Elastic entity) throws IOException {
            IndexRequest request = new IndexRequest(idxName);
            request.id(entity.getId());
            request.source(JSON.toJSONString(entity.getData()), XContentType.JSON);
            restHighLevelClient.index(request, RequestOptions.DEFAULT);
        }
    
        /**
         * @since
         */
        public void deleteOne(String idxName, Elastic entity) {
            DeleteRequest request = new DeleteRequest(idxName);
            request.id(entity.getId());
            try {
                restHighLevelClient.delete(request,RequestOptions.DEFAULT);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    
        /** 批量插入数据
         * @since
         */
        public void insertBatch(String idxName, List<Elastic> list) throws IOException {
            BulkRequest request = new BulkRequest();
            list.forEach(item -> request.add(new IndexRequest(idxName).id(item.getId())
                    .source(JSON.toJSONString(item.getData()), XContentType.JSON)));
            restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
        }
    
        /** 批量插入数据
         * @throws
         * @since
         */
        public void insertBatchTrueObj(String idxName, List<Elastic> list) throws IOException {
            BulkRequest request = new BulkRequest();
            list.forEach(item -> request.add(new IndexRequest(idxName).id(item.getId())
                    .source(item.getData(), XContentType.JSON)));
            restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
        }
    
    
        /**
         * 分页搜索
         * @author xzr
         * @param idxName
         * @param builder
         * @param currPage
         * @param pageSize
         * @return
         * @throws IOException
         */
        public Page<Map<String, Object>> search(String idxName, SearchSourceBuilder builder, int currPage, int pageSize) throws IOException {
        	//设置分页
        	builder.from((currPage - 1) * pageSize);
        	builder.size(pageSize);
        	//设置搜索的超时时间
        	builder.timeout(new TimeValue(60, TimeUnit.SECONDS)); 
        	SearchRequest request = new SearchRequest(idxName);
            request.source(builder);
            SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
            SearchHit[] hits = response.getHits().getHits();
            
            int totalCount = (int) response.getHits().getTotalHits();
            List<Map<String, Object>> list = new ArrayList<>(hits.length);
            for (SearchHit hit : hits) {
                list.add(hit.getSourceAsMap());
            }
            return new Page<Map<String, Object>>(list, totalCount, pageSize, currPage);
        }
        
        /**
         * 高亮分页搜索显示
         * @author xzr
         * @param idxName
         * @param builder
         * @param currPage
         * @param pageSize
         * @return
         * @throws IOException
         */
        public Page<Map<String, Object>> search(String idxName, SearchSourceBuilder builder, int currPage, int pageSize, String... highLightFieldNames) throws IOException {
        	//设置分页
        	builder.from((currPage - 1) * pageSize);
        	builder.size(pageSize);
        	//设置搜索的超时时间
        	builder.timeout(new TimeValue(60, TimeUnit.SECONDS)); 
        	//设置高亮显示
            HighlightBuilder highlightBuilder = new HighlightBuilder();
            //.field("*").requireFieldMatch(false);
            for(String field : highLightFieldNames) {
            	highlightBuilder.field(field).requireFieldMatch(true);
            }
            highlightBuilder.preTags("<font style=\"color:red\">");
            highlightBuilder.postTags("</font>");
            builder.highlighter(highlightBuilder);
     
            SearchRequest searchRequest = new SearchRequest(idxName);
            searchRequest.source(builder);
            //System.out.println("builder:" + builder);
            SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
     
            //遍历结果
            for(SearchHit hit : response.getHits()){
                Map<String, Object> source = hit.getSourceAsMap();
                //处理高亮片段
                Map<String, HighlightField> highlightFields = hit.getHighlightFields();
                for (String highLightFieldName : highLightFieldNames) {
                	HighlightField nameField = highlightFields.get(highLightFieldName);
                    if(null != nameField){
                        Text[] fragments = nameField.fragments();
                        StringBuilder nameTmp = new StringBuilder();
                        for(Text text : fragments){
                            nameTmp.append(text);
                        }
                        //将高亮片段组装到结果中去
                        source.put(highLightFieldName, nameTmp.toString());
                    }
                }
            }
            int totalCount = (int) response.getHits().getTotalHits();
            SearchHit[] hits = response.getHits().getHits();
            List<Map<String, Object>> list = new ArrayList<>(hits.length);
            for (SearchHit hit : hits) {
            	list.add(hit.getSourceAsMap());
            }
            return new Page<Map<String, Object>>(list, totalCount, pageSize, currPage);
        }
    
    
        /**
         * 分页搜索
         * @author xzr
         * @param idxName
         * @param builder
         * @param from
         * @param size
         * @return
         * @throws IOException
         */
        public Set<String> searchForSet(String idxName, SearchSourceBuilder builder, int from, int size) throws IOException {
            //设置分页
            builder.from(from);
            builder.size(size);
            SearchRequest request = new SearchRequest(idxName);
            request.source(builder);
    
            SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
            SearchHit[] hits = response.getHits().getHits();
            Set<String> hashSet=new LinkedHashSet<>();
            hashSet = Stream.of(hits)
                              .map(hit -> String.valueOf(hit.getSourceAsMap().get("id")))
                              .collect(Collectors.toSet());
            return hashSet;
        }
    
        public Set<String> searchForSet(String idxName, SearchSourceBuilder builder, int from, int size,String field) throws IOException {
            //设置分页
            builder.from(from);
            builder.size(size);
            SearchRequest request = new SearchRequest(idxName);
            request.source(builder);
    
            SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
            SearchHit[] hits = response.getHits().getHits();
            Set<String> hashSet=new LinkedHashSet<>();
            hashSet = Stream.of(hits)
                            .map(hit -> String.valueOf(hit.getSourceAsMap().get(field)))
                            .collect(Collectors.toSet());
            return hashSet;
        }
    
        /**
         * 不分页搜索
         * @param idxName
         * @param builder
         * @return
         * @throws IOException
         */
        public List<Map<String, Object>> search(String idxName, SearchSourceBuilder builder) throws IOException {
            //设置分页
            builder.from(1);
            //阿里云ES默认最大允许单次10000条
            builder.size(200000);
            //设置搜索的超时时间
            builder.timeout(new TimeValue(60, TimeUnit.SECONDS));
            SearchRequest request = new SearchRequest(idxName);
            request.source(builder);
            SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
            SearchHit[] hits = response.getHits().getHits();
    
            List<Map<String, Object>> list = new ArrayList<>(hits.length);
            for (SearchHit hit : hits) {
                list.add(hit.getSourceAsMap());
            }
            return list;
        }
    
        public String getScrollId(String idxName, SearchSourceBuilder builder, int size) throws IOException {
            builder.size(size);
            //设置搜索的超时时间
            builder.timeout(TimeValue.timeValueMinutes(5));
            SearchRequest request = new SearchRequest(idxName);
            request.scroll(new Scroll(TimeValue.timeValueMinutes(5)));
            request.source(builder);
            SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
            return response.getScrollId();
        }
    
    
        /**
         *@Description: 通过scrollId查询
         */
        public SearchResponse searchByScrollId(String scrollId) throws IOException {
    
            SearchScrollRequest searchScrollRequest = new SearchScrollRequest();
            searchScrollRequest.scrollId(scrollId);
            searchScrollRequest.scroll(TimeValue.timeValueMinutes(5));
            return restHighLevelClient.scroll(searchScrollRequest, RequestOptions.DEFAULT);
    
        }
    
    
    
    
        /**
         * @param idxName
         * @param builder
         */
        public Long searchCount(String idxName, SearchSourceBuilder builder) throws IOException {
            //设置搜索的超时时间
            CountRequest request = new CountRequest(idxName);
            request.source(builder);
            CountResponse response = restHighLevelClient.count(request, RequestOptions.DEFAULT);
            return response.getCount();
        }
    
    
    
        /**
         * 按字段分组搜索
         */
        public Set<String> searchForSetByGroup(String idxName, SearchSourceBuilder builder, String groupField) throws IOException {
            //设置分页
            builder.from(0);
            builder.size(1);
            //设置搜索的超时时间
            builder.timeout(new TimeValue(60, TimeUnit.SECONDS));
            SearchRequest request = new SearchRequest(idxName);
            request.source(builder);
            SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
            Aggregations agg = response.getAggregations();
            Terms terms = agg.get(groupField);
            Set<String> collect = terms.getBuckets().stream().map(bucket -> String.valueOf(bucket.getKey()))
                                       .collect(Collectors.toSet());
            return collect;
        }
    
    
    
        /** 删除index
         * @since
         */
        public void deleteIndex(String idxName) throws Exception {
            if (!this.indexExist(idxName)) {
                return;
            }
            restHighLevelClient.indices().delete(new DeleteIndexRequest(idxName), RequestOptions.DEFAULT);
        }
    
    
        /**
         * @throws
         * @since
         */
        public void deleteByQuery(String idxName, QueryBuilder builder) throws IOException {
    
            DeleteByQueryRequest request = new DeleteByQueryRequest(idxName);
            request.setQuery(builder);
            //设置批量操作数量,最大为10000
            request.setBatchSize(10000);
            request.setConflicts("proceed");
            restHighLevelClient.deleteByQuery(request, RequestOptions.DEFAULT);
        }
    }

logstash同步数据到es:

input {
      jdbc {
    	jdbc_driver_library => "E:\mysql-connector-java-5.1.46\mysql-connector-java-5.1.46\mysql-connector-java-5.1.46.jar"
    	jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_connection_string => "数据库地址"
        jdbc_user => root
        jdbc_password => "密码"
    	jdbc_paging_enabled => "true"
        jdbc_page_size => "3000"
    	jdbc_default_timezone => "Asia/Shanghai"
        statement => "查询数据的sql"
        record_last_run => true
        tracking_column_type => "numeric"
        tracking_column => "update_time"
        use_column_value => true
        last_run_metadata_path => "D:\bid_last_id.txt"
      }
    }
    
    filter {
      mutate{
        copy => { "bid" => "[@metadata][_id]"}
        remove_field => ["@version","@timestamp"]
      }
    }
    
    output {
      elasticsearch {
        index => "jsk_search_bid"
        hosts => ["192.168.101.101:9200"]
    	doc_as_upsert => true
        action => "update"
    	document_id => "%{[@metadata][_id]}"
      }
    
    }