中间件：ElasticSearch组件RestHighLevelClient用法详解

2021-01-25 549

一、基础API简介

1、RestHighLevelClient

RestHighLevelClient的API作为ElasticSearch备受推荐的客户端组件，其封装系统操作ES的方法，包括索引结构管理，数据增删改查管理，常用查询方法，并且可以结合原生ES查询原生语法，功能十分强大。

在使用RestHighLevelClient的语法时，通常涉及上面几个方面，在掌握基础用法之上可以根据业务特点进行一些自定义封装，这样可以更优雅的解决业务需求。

2、核心依赖

使用RestHighLevelClient需要依赖rest-high-level-client包，和ES相关基础依赖。

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

二、索引管理

这里不做过多描述，注意一点：因为ES的数据结构特点，所以不需要索引更新方法，新的字段在更新数据时直接写入即可，不需要提前更新索引结构。

@Service
public class EsIndexOperation {

    @Resource
    private RestHighLevelClient client ;
    private final RequestOptions options = RequestOptions.DEFAULT;

    /**
     * 判断索引是否存在
     */
    public boolean checkIndex (String index) {
        try {
            return client.indices().exists(new GetIndexRequest(index), options);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return Boolean.FALSE ;
    }

    /**
     * 创建索引
     */
    public boolean createIndex (String indexName ,Map<String, Object> columnMap){
        try {
            if(!checkIndex(indexName)){
                CreateIndexRequest request = new CreateIndexRequest(indexName);
                if (columnMap != null && columnMap.size()>0) {
                    Map<String, Object> source = new HashMap<>();
                    source.put("properties", columnMap);
                    request.mapping(source);
                }
                this.client.indices().create(request, options);
                return Boolean.TRUE ;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return Boolean.FALSE;
    }

    /**
     * 删除索引
     */
    public boolean deleteIndex(String indexName) {
        try {
            if(checkIndex(indexName)){
                DeleteIndexRequest request = new DeleteIndexRequest(indexName);
                AcknowledgedResponse response = client.indices().delete(request, options);
                return response.isAcknowledged();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return Boolean.FALSE;
    }
}

三、数据管理

这里在更新数据时，可以直接修改索引结构，在dataMap中放进新的字段即可。

@Service
public class EsDataOperation {

    @Resource
    private RestHighLevelClient client ;
    private final RequestOptions options = RequestOptions.DEFAULT;

    /**
     * 写入数据
     */
    public boolean insert (String indexName, Map<String,Object> dataMap){
        try {
            BulkRequest request = new BulkRequest();
            request.add(new IndexRequest(indexName,"doc").id(dataMap.remove("id").toString())
                    .opType("create").source(dataMap,XContentType.JSON));
            this.client.bulk(request, options);
            return Boolean.TRUE ;
        } catch (Exception e){
            e.printStackTrace();
        }
        return Boolean.FALSE;
    }

    /**
     * 批量写入数据
     */
    public boolean batchInsert (String indexName, List<Map<String,Object>> userIndexList){
        try {
            BulkRequest request = new BulkRequest();
            for (Map<String,Object> dataMap:userIndexList){
                request.add(new IndexRequest(indexName,"doc").id(dataMap.remove("id").toString())
                        .opType("create").source(dataMap,XContentType.JSON));
            }
            this.client.bulk(request, options);
            return Boolean.TRUE ;
        } catch (Exception e){
            e.printStackTrace();
        }
        return Boolean.FALSE;
    }

    /**
     * 更新数据，可以直接修改索引结构
     */
    public boolean update (String indexName, Map<String,Object> dataMap){
        try {
            UpdateRequest updateRequest = new UpdateRequest(indexName,"doc", dataMap.remove("id").toString());
            updateRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
            updateRequest.doc(dataMap) ;
            this.client.update(updateRequest, options);
            return Boolean.TRUE ;
        } catch (Exception e){
            e.printStackTrace();
        }
        return Boolean.FALSE;
    }

    /**
     * 删除数据
     */
    public boolean delete (String indexName, String id){
        try {
            DeleteRequest deleteRequest = new DeleteRequest(indexName,"doc", id);
            this.client.delete(deleteRequest, options);
            return Boolean.TRUE ;
        } catch (Exception e){
            e.printStackTrace();
        }
        return Boolean.FALSE;
    }
}

四、查询操作

注意：查询总数的CountRequest语法，SearchRequest查询结果中数据转换语法，分页查询中需要指定偏移位置和分页大小。

@Service
public class EsQueryOperation {

    @Resource
    private RestHighLevelClient client ;
    private final RequestOptions options = RequestOptions.DEFAULT;

    /**
     * 查询总数
     */
    public Long count (String indexName){
        // 指定创建时间
        BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
        queryBuilder.must(QueryBuilders.termQuery("createTime", 1611378102795L));

        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(queryBuilder);

        CountRequest countRequest = new CountRequest(indexName);
        countRequest.source(sourceBuilder);
        try {
            CountResponse countResponse = client.count(countRequest, options);
            return countResponse.getCount();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return 0L;
    }

    /**
     * 查询集合
     */
    public List<Map<String,Object>> list (String indexName) {
        // 查询条件,指定时间并过滤指定字段值
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
        queryBuilder.must(QueryBuilders.termQuery("createTime", 1611378102795L));
        queryBuilder.mustNot(QueryBuilders.termQuery("name","北京-李四"));
        sourceBuilder.query(queryBuilder);
        SearchRequest searchRequest = new SearchRequest(indexName);
        searchRequest.source(sourceBuilder);
        try {
            SearchResponse searchResp = client.search(searchRequest, options);
            List<Map<String,Object>> data = new ArrayList<>() ;
            SearchHit[] searchHitArr = searchResp.getHits().getHits();
            for (SearchHit searchHit:searchHitArr){
                Map<String,Object> temp = searchHit.getSourceAsMap();
                temp.put("id",searchHit.getId()) ;
                data.add(temp);
            }
            return data;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null ;
    }

    /**
     * 分页查询
     */
    public List<Map<String,Object>> page (String indexName,Integer offset,Integer size) {
        // 查询条件,指定时间并过滤指定字段值
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.from(offset);
        sourceBuilder.size(size);
        sourceBuilder.sort("createTime", SortOrder.DESC);
        SearchRequest searchRequest = new SearchRequest(indexName);
        searchRequest.source(sourceBuilder);
        try {
            SearchResponse searchResp = client.search(searchRequest, options);
            List<Map<String,Object>> data = new ArrayList<>() ;
            SearchHit[] searchHitArr = searchResp.getHits().getHits();
            for (SearchHit searchHit:searchHitArr){
                Map<String,Object> temp = searchHit.getSourceAsMap();
                temp.put("id",searchHit.getId()) ;
                data.add(temp);
            }
            return data;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null ;
    }
}

五、排序方式

排序除了常规的指定字段升序降序规则之外，还可以基于原生的脚本语法，基于自定义规则排序让一些特定的数据沉底或者置顶。

@Service
public class EsSortOperation {

    @Resource
    private RestHighLevelClient client ;
    private final RequestOptions options = RequestOptions.DEFAULT;

    /**
     * 排序规则
     */
    public List<Map<String,Object>> sort (String indexName) {
        // 先升序时间，在倒序年龄
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.sort("createTime", SortOrder.ASC);
        sourceBuilder.sort("age",SortOrder.DESC) ;
        SearchRequest searchRequest = new SearchRequest(indexName);
        searchRequest.source(sourceBuilder);
        try {
            SearchResponse searchResp = client.search(searchRequest, options);
            List<Map<String,Object>> data = new ArrayList<>() ;
            SearchHit[] searchHitArr = searchResp.getHits().getHits();
            for (SearchHit searchHit:searchHitArr){
                Map<String,Object> temp = searchHit.getSourceAsMap();
                temp.put("id",searchHit.getId()) ;
                data.add(temp);
            }
            return data;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null ;
    }

    /**
     * 自定义排序规则
     */
    public List<Map<String,Object>> defSort (String indexName) {
        // 指定置换顺序的规则
        // [age 12-->60]\[age 19-->10]\[age 13-->30]\[age 18-->40],age其他值忽略为1
        Script script = new Script("def _ageSort = doc['age'].value == 12?60:" +
                                                            "(doc['age'].value == 19?10:" +
                                                            "(doc['age'].value == 13?30:" +
                                                            "(doc['age'].value == 18?40:1)));" + "_ageSort;");
        ScriptSortBuilder sortBuilder = SortBuilders.scriptSort(script,ScriptSortBuilder.ScriptSortType.NUMBER);
        sortBuilder.order(SortOrder.ASC);
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.sort(sortBuilder);
        SearchRequest searchRequest = new SearchRequest(indexName);
        searchRequest.source(sourceBuilder);
        try {
            SearchResponse searchResp = client.search(searchRequest, options);
            List<Map<String,Object>> data = new ArrayList<>() ;
            SearchHit[] searchHitArr = searchResp.getHits().getHits();
            for (SearchHit searchHit:searchHitArr){
                Map<String,Object> temp = searchHit.getSourceAsMap();
                temp.put("id",searchHit.getId()) ;
                data.add(temp);
            }
            return data;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null ;
    }
}

六、源代码地址

GitHub·地址
https://github.com/cicadasmile/data-manage-parent
GitEE·地址
https://gitee.com/cicadasmile/data-manage-parent

微信关注我们

原文链接：https://my.oschina.net/cicadasmile/blog/4922681

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

SAP UI渲染模式：客户端渲染 VS 服务器端渲染

今年由于疫情原因，SAP一年一度的DKOM(Developer Kick-Off Meeting)切换成了线上模式，刚刚于昨天圆满落幕。时间过得真快，去年的DKOM仿佛还历历在目：SAP成都研究院数字创新空间小伙伴们在2020 SAP上海DKOM. 但是Jerry昨天没有太多时间去在线观看，只因为… 我们团队昨天收到了一个优先级为VERY HIGH的客户incident：这个incident是客户在使用SAP Spartacus并试图启用其服务器端渲染模式(Server Side Rendering，简称SSR)时，遇到了问题，因此我有机会回顾之前使用过的SAP UI开发技术里，关于客户端渲染(Client Side Rendering，简称CSR)和服务器端渲染的知识点。回顾目录 SAP ABAP Webdynpro - 服务器端渲染 SAP BSP / WebClient UI - 服务器端渲染 SAP UI5 - 客户端渲染 SAP Commerce Cloud Accelerator UI - 服务器端渲染 Jerry手头正在处理的incident - SAP * Spa...

2021-01-24

636

在处理实时流数据时，不仅需要按照时间做纵向聚合计算（时间序列聚合引擎），还需要对最新的数据做横向比较和计算，如金融里对所有股票的最新报价求百分位、工业物联网中计算一批设备的温度均值等。DolphinDB database 提供了横截面聚合引擎，可以对流数据中所有分组的最新数据做聚合运算。横截面引擎的主体分为两部分：横截面数据表和计算引擎。横截面数据是横截面引擎的内部表，保存了所有分组最新的截面数据。计算引擎是一组聚合计算表达式以及触发器，系统会按照指定的方式触发聚合运算，计算结果会输出到另外一个表中。 1. 基本用法在DolphinDB database中，通过createCrossSectionalAggregator创建横截面聚合引擎。它返回一个横截面数据表，保存了所有分组最新的截面数据，往这个表写入数据意味着这些数据进入横截面聚合引擎进行计算。具体用法如下： createCrossSectionalAggregator(name, [metrics], dummyTable, [outputTable], keyColumn, [triggeringPattern="perB...

2021-01-25

633

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。