Elasticsearch 系列文章

1、介紹lucene的功能以及建立索引、搜索單詞、搜索詞語(yǔ)和搜索句子四個(gè)示例實(shí)現(xiàn)
2、Elasticsearch7.6.1基本介紹、2種部署方式及驗(yàn)證、head插件安裝、分詞器安裝及驗(yàn)證
3、Elasticsearch7.6.1信息搜索示例（索引操作、數(shù)據(jù)操作-添加、刪除、導(dǎo)入等、數(shù)據(jù)搜索及分頁(yè)）
4、Elasticsearch7.6.1 Java api操作ES（CRUD、兩種分頁(yè)方式、高亮顯示）和Elasticsearch SQL詳細(xì)示例
5、Elasticsearch7.6.1 filebeat介紹及收集kafka日志到es示例
6、Elasticsearch7.6.1、logstash、kibana介紹及綜合示例（ELK、grok插件）
7、Elasticsearch7.6.1收集nginx日志及監(jiān)測(cè)指標(biāo)示例
8、Elasticsearch7.6.1收集mysql慢查詢?nèi)罩炯氨O(jiān)控
9、Elasticsearch7.6.1 ES與HDFS相互轉(zhuǎn)存數(shù)據(jù)-ES-Hadoop

本文簡(jiǎn)單的介紹了java api操作Elasticsearch和Elasticsearch SQL的詳細(xì)示例。
本文依賴es環(huán)境可用。
本文分為2個(gè)部分，及java api操作es和es sql的詳細(xì)使用示例。

一、java api操作ES

使用一個(gè)JobService類來(lái)實(shí)現(xiàn)上一篇文章中的示例，此處用RESTFul完成的操作
需要增加一個(gè)日志文件配置
官網(wǎng)API地址：https://www.elastic.co/guide/en/elasticsearch/client/java-rest/7.6/java-rest-high.html

1、pom.xml

<dependencies>
        <!-- ES的高階的客戶端API -->
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-high-level-client</artifactId>
            <version>7.6.1</version>
        </dependency>

        <!-- 阿里巴巴出品的一款將Java對(duì)象轉(zhuǎn)換為JSON、將JSON轉(zhuǎn)換為Java對(duì)象的庫(kù) -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.62</version>
        </dependency>

        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>2.11.1</version>
        </dependency>
    </dependencies>

2、日志配置

3、bean

在id字段上添加一個(gè) @JSONField注解，并配置注解的serialize為false，表示該字段無(wú)需轉(zhuǎn)換為JSON，因?yàn)樗褪俏臋n的唯一ID。

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.annotation.JSONField;

import lombok.Data;

/**
 * @author chenw
 *
 */
@Data
public class JobDetail {
    @JSONField(serialize = false)
    private long id;
    private String area;
    private String exp;
    private String edu;
    private String salary;
    private String job_type;
    private String cmp;
    private String pv;
    private String title;
    private String jd;

    @Override
    public String toString() {
        // 使用FastJSON將一個(gè)對(duì)象直接轉(zhuǎn)換為JSON字符串
        return id + ":" + JSONObject.toJSONString(this);
    }
}

4、service及實(shí)現(xiàn)

• 服務(wù)接口

import java.io.IOException;
import java.util.List;
import java.util.Map;

import org.example.bean.JobDetail;

/**
 * @author chenw
 *
 */
public interface JobFullTextService {
    // 添加一個(gè)職位數(shù)據(jù)
    void add(JobDetail jobDetail) throws IOException;

    // 根據(jù)ID檢索指定職位數(shù)據(jù)
    JobDetail findById(long id) throws IOException;

    // 修改職位薪資
    void update(JobDetail jobDetail) throws IOException;

    // 根據(jù)ID刪除指定位置數(shù)據(jù)
    void deleteById(long id) throws IOException;

    // 根據(jù)關(guān)鍵字檢索數(shù)據(jù)
    List<JobDetail> searchByKeywords(String keywords) throws IOException;

    // 分頁(yè)檢索
    Map<String, Object> searchByPage(String keywords, int pageNum, int pageSize) throws IOException;

    // scroll分頁(yè)解決深分頁(yè)問(wèn)題
    Map<String, Object> searchByScrollPage(String keywords, String scrollId, int pageSize) throws IOException;

    // 關(guān)閉ES連接
    void close() throws IOException;
}

• 接口實(shí)現(xiàn)

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.http.HttpHost;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.MultiMatchQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.example.bean.JobDetail;
import org.example.service.JobFullTextService;

import com.alibaba.fastjson.JSONObject;

/**
 * @author chenw
 *
 */
public class JobFullTextServiceImpl implements JobFullTextService {
    // 索引庫(kù)的名字
    private static final String JOB_IDX = "job_idx";
    private RestHighLevelClient restHighLevelClient;

    public JobFullTextServiceImpl() {
        // 建立與ES的連接
        // 1. 使用RestHighLevelClient構(gòu)建客戶端連接。
        // 2. 基于RestClient.builder方法來(lái)構(gòu)建RestClientBuilder
        // 3. 用HttpHost來(lái)添加ES的節(jié)點(diǎn)
        RestClientBuilder restClientBuilder = RestClient.builder(
                new HttpHost("server1", 9200, "http"), 
                new HttpHost("server2", 9200, "http"),
                new HttpHost("server3", 9200, "http"));
        restHighLevelClient = new RestHighLevelClient(restClientBuilder);
    }

    /**
     * @param args
     */
    public static void main(String[] args) {

    }

    @Override
    public void add(JobDetail jobDetail) throws IOException {
        // 1. 構(gòu)建IndexRequest對(duì)象，用來(lái)描述ES發(fā)起請(qǐng)求的數(shù)據(jù)。
        IndexRequest indexRequest = new IndexRequest(JOB_IDX);

        // 2. 設(shè)置文檔ID。
        indexRequest.id(jobDetail.getId() + "");

        // 3. 使用FastJSON將實(shí)體類對(duì)象轉(zhuǎn)換為JSON。
        String json = JSONObject.toJSONString(jobDetail);

        // 4. 使用IndexRequest.source方法設(shè)置文檔數(shù)據(jù)，并設(shè)置請(qǐng)求的數(shù)據(jù)為JSON格式。
        indexRequest.source(json, XContentType.JSON);

        // 5. 使用ES High level client調(diào)用index方法發(fā)起請(qǐng)求，將一個(gè)文檔添加到索引中。
        restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
    }

    @Override
    public JobDetail findById(long id) throws IOException {
        // 1. 構(gòu)建GetRequest請(qǐng)求。
        GetRequest getRequest = new GetRequest(JOB_IDX, id + "");

        // 2. 使用RestHighLevelClient.get發(fā)送GetRequest請(qǐng)求，并獲取到ES服務(wù)器的響應(yīng)。
        GetResponse getResponse = restHighLevelClient.get(getRequest, RequestOptions.DEFAULT);

        // 3. 將ES響應(yīng)的數(shù)據(jù)轉(zhuǎn)換為JSON字符串
        String json = getResponse.getSourceAsString();

        // 4. 并使用FastJSON將JSON字符串轉(zhuǎn)換為JobDetail類對(duì)象
        JobDetail jobDetail = JSONObject.parseObject(json, JobDetail.class);

        // 5. 記得：?jiǎn)为?dú)設(shè)置ID
        jobDetail.setId(id);

        return jobDetail;
    }

    @Override
    public void update(JobDetail jobDetail) throws IOException {
        // 1. 判斷對(duì)應(yīng)ID的文檔是否存在
        // a) 構(gòu)建GetRequest
        GetRequest getRequest = new GetRequest(JOB_IDX, jobDetail.getId() + "");

        // b) 執(zhí)行client的exists方法，發(fā)起請(qǐng)求，判斷是否存在
        boolean exists = restHighLevelClient.exists(getRequest, RequestOptions.DEFAULT);

        if (exists) {
            // 2. 構(gòu)建UpdateRequest請(qǐng)求
            UpdateRequest updateRequest = new UpdateRequest(JOB_IDX, jobDetail.getId() + "");

            // 3. 設(shè)置UpdateRequest的文檔，并配置為JSON格式
            updateRequest.doc(JSONObject.toJSONString(jobDetail), XContentType.JSON);

            // 4. 執(zhí)行client發(fā)起update請(qǐng)求
            restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT);
        }

    }

    @Override
    public void deleteById(long id) throws IOException {
        // 1. 構(gòu)建delete請(qǐng)求
        DeleteRequest deleteRequest = new DeleteRequest(JOB_IDX, id + "");

        // 2. 使用RestHighLevelClient執(zhí)行delete請(qǐng)求
        restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);

    }

    @Override
    public List<JobDetail> searchByKeywords(String keywords) throws IOException {
        /// 1.構(gòu)建SearchRequest檢索請(qǐng)求
        // 專門用來(lái)進(jìn)行全文檢索、關(guān)鍵字檢索的API
        SearchRequest searchRequest = new SearchRequest(JOB_IDX);

        // 2.創(chuàng)建一個(gè)SearchSourceBuilder專門用于構(gòu)建查詢條件
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();

        // 3.使用QueryBuilders.multiMatchQuery構(gòu)建一個(gè)查詢條件（搜索title、jd），并配置到SearchSourceBuilder
        MultiMatchQueryBuilder multiMatchQueryBuilder = QueryBuilders.multiMatchQuery(keywords, "title", "jd");

        // 將查詢條件設(shè)置到查詢請(qǐng)求構(gòu)建器中
        searchSourceBuilder.query(multiMatchQueryBuilder);

        // 4.調(diào)用SearchRequest.source將查詢條件設(shè)置到檢索請(qǐng)求
        searchRequest.source(searchSourceBuilder);

        // 5.執(zhí)行RestHighLevelClient.search發(fā)起請(qǐng)求
        SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        SearchHit[] hitArray = searchResponse.getHits().getHits();

        // 6.遍歷結(jié)果
        ArrayList<JobDetail> jobDetailArrayList = new ArrayList<>();

        for (SearchHit documentFields : hitArray) {
            // 1)獲取命中的結(jié)果
            String json = documentFields.getSourceAsString();

            // 2)將JSON字符串轉(zhuǎn)換為對(duì)象
            JobDetail jobDetail = JSONObject.parseObject(json, JobDetail.class);

            // 3)使用SearchHit.getId設(shè)置文檔ID
            jobDetail.setId(Long.parseLong(documentFields.getId()));

            jobDetailArrayList.add(jobDetail);
        }

        return jobDetailArrayList;
    }

    @Override
    public Map<String, Object> searchByPage(String keywords, int pageNum, int pageSize) throws IOException {
        // 1.構(gòu)建SearchRequest檢索請(qǐng)求
        // 專門用來(lái)進(jìn)行全文檢索、關(guān)鍵字檢索的API
        SearchRequest searchRequest = new SearchRequest(JOB_IDX);

        // 2.創(chuàng)建一個(gè)SearchSourceBuilder專門用于構(gòu)建查詢條件
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();

        // 3.使用QueryBuilders.multiMatchQuery構(gòu)建一個(gè)查詢條件（搜索title、jd），并配置到SearchSourceBuilder
        MultiMatchQueryBuilder multiMatchQueryBuilder = QueryBuilders.multiMatchQuery(keywords, "title", "jd");

        // 將查詢條件設(shè)置到查詢請(qǐng)求構(gòu)建器中
        searchSourceBuilder.query(multiMatchQueryBuilder);

        // 每頁(yè)顯示多少條
        searchSourceBuilder.size(pageSize);
        // 設(shè)置從第幾條開(kāi)始查詢
        searchSourceBuilder.from((pageNum - 1) * pageSize);

        // 4.調(diào)用SearchRequest.source將查詢條件設(shè)置到檢索請(qǐng)求
        searchRequest.source(searchSourceBuilder);

        // 5.執(zhí)行RestHighLevelClient.search發(fā)起請(qǐng)求
        SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        SearchHit[] hitArray = searchResponse.getHits().getHits();

        // 6.遍歷結(jié)果
        ArrayList<JobDetail> jobDetailArrayList = new ArrayList<>();

        for (SearchHit documentFields : hitArray) {
            // 1)獲取命中的結(jié)果
            String json = documentFields.getSourceAsString();

            // 2)將JSON字符串轉(zhuǎn)換為對(duì)象
            JobDetail jobDetail = JSONObject.parseObject(json, JobDetail.class);

            // 3)使用SearchHit.getId設(shè)置文檔ID
            jobDetail.setId(Long.parseLong(documentFields.getId()));

            jobDetailArrayList.add(jobDetail);
        }

        // 8. 將結(jié)果封裝到Map結(jié)構(gòu)中（帶有分頁(yè)信息）
        // a) total -> 使用SearchHits.getTotalHits().value獲取到所有的記錄數(shù)
        // b) content -> 當(dāng)前分頁(yè)中的數(shù)據(jù)
        long totalNum = searchResponse.getHits().getTotalHits().value;
        HashMap hashMap = new HashMap();
        hashMap.put("total", totalNum);
        hashMap.put("content", jobDetailArrayList);

        return hashMap;
    }

    @Override
    public Map<String, Object> searchByScrollPage(String keywords, String scrollId, int pageSize) throws IOException {
        SearchResponse searchResponse = null;

        if (scrollId == null) {
            // 1.構(gòu)建SearchRequest檢索請(qǐng)求
            // 專門用來(lái)進(jìn)行全文檢索、關(guān)鍵字檢索的API
            SearchRequest searchRequest = new SearchRequest(JOB_IDX);

            // 2.創(chuàng)建一個(gè)SearchSourceBuilder專門用于構(gòu)建查詢條件
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();

            // 3.使用QueryBuilders.multiMatchQuery構(gòu)建一個(gè)查詢條件（搜索title、jd），并配置到SearchSourceBuilder
            MultiMatchQueryBuilder multiMatchQueryBuilder = QueryBuilders.multiMatchQuery(keywords, "title", "jd");

            // 將查詢條件設(shè)置到查詢請(qǐng)求構(gòu)建器中
            searchSourceBuilder.query(multiMatchQueryBuilder);

            // 設(shè)置高亮
            HighlightBuilder highlightBuilder = new HighlightBuilder();
            highlightBuilder.field("title");
            highlightBuilder.field("jd");
            highlightBuilder.preTags("<font color='red'>");
            highlightBuilder.postTags("</font>");

            // 給請(qǐng)求設(shè)置高亮
            searchSourceBuilder.highlighter(highlightBuilder);

            // 每頁(yè)顯示多少條
            searchSourceBuilder.size(pageSize);

            // 4.調(diào)用SearchRequest.source將查詢條件設(shè)置到檢索請(qǐng)求
            searchRequest.source(searchSourceBuilder);

            // --------------------------
            // 設(shè)置scroll查詢
            // --------------------------
            searchRequest.scroll(TimeValue.timeValueMinutes(5));

            // 5.執(zhí)行RestHighLevelClient.search發(fā)起請(qǐng)求
            searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);

        }
        // 第二次查詢的時(shí)候，直接通過(guò)scroll id查詢數(shù)據(jù)
        else {
            SearchScrollRequest searchScrollRequest = new SearchScrollRequest(scrollId);
            searchScrollRequest.scroll(TimeValue.timeValueMinutes(5));

            // 使用RestHighLevelClient發(fā)送scroll請(qǐng)求
            searchResponse = restHighLevelClient.scroll(searchScrollRequest, RequestOptions.DEFAULT);
        }

        // --------------------------
        // 迭代ES響應(yīng)的數(shù)據(jù)
        // --------------------------

        SearchHit[] hitArray = searchResponse.getHits().getHits();

        // 6.遍歷結(jié)果
        ArrayList<JobDetail> jobDetailArrayList = new ArrayList<>();

        for (SearchHit documentFields : hitArray) {
            // 1)獲取命中的結(jié)果
            String json = documentFields.getSourceAsString();

            // 2)將JSON字符串轉(zhuǎn)換為對(duì)象
            JobDetail jobDetail = JSONObject.parseObject(json, JobDetail.class);

            // 3)使用SearchHit.getId設(shè)置文檔ID
            jobDetail.setId(Long.parseLong(documentFields.getId()));

            jobDetailArrayList.add(jobDetail);

            // 設(shè)置高亮的一些文本到實(shí)體類中
            // 封裝了高亮
            Map<String, HighlightField> highlightFieldMap = documentFields.getHighlightFields();
            HighlightField titleHL = highlightFieldMap.get("title");
            HighlightField jdHL = highlightFieldMap.get("jd");

            if (titleHL != null) {
                // 獲取指定字段的高亮片段
                Text[] fragments = titleHL.getFragments();
                // 將這些高亮片段拼接成一個(gè)完整的高亮字段
                StringBuilder builder = new StringBuilder();
                for (Text text : fragments) {
                    builder.append(text);
                }
                // 設(shè)置到實(shí)體類中
                jobDetail.setTitle(builder.toString());
            }

            if (jdHL != null) {
                // 獲取指定字段的高亮片段
                Text[] fragments = jdHL.getFragments();
                // 將這些高亮片段拼接成一個(gè)完整的高亮字段
                StringBuilder builder = new StringBuilder();
                for (Text text : fragments) {
                    builder.append(text);
                }
                // 設(shè)置到實(shí)體類中
                jobDetail.setJd(builder.toString());
            }
        }

        // 8. 將結(jié)果封裝到Map結(jié)構(gòu)中（帶有分頁(yè)信息）
        // a) total -> 使用SearchHits.getTotalHits().value獲取到所有的記錄數(shù)
        // b) content -> 當(dāng)前分頁(yè)中的數(shù)據(jù)
        long totalNum = searchResponse.getHits().getTotalHits().value;
        HashMap hashMap = new HashMap();
        hashMap.put("scroll_id", searchResponse.getScrollId());
        hashMap.put("content", jobDetailArrayList);

        return hashMap;
    }

    @Override
    public void close() throws IOException {
        restHighLevelClient.close();
    }

}

5、驗(yàn)證

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.example.bean.JobDetail;
import org.example.service.JobFullTextService;
import org.example.service.impl.JobFullTextServiceImpl;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

/**
 * @author chenw
 *
 */
public class JobFullTextServiceTest {
    private JobFullTextService jobFullTextService;

    @Before
    public void beforeTest() {
        jobFullTextService = new JobFullTextServiceImpl();
    }

    @Test
    public void addTest() throws IOException {
        JobDetail jobDetail = new JobDetail();
        jobDetail.setId(1);
        jobDetail.setArea("江蘇省-南京市");
        jobDetail.setCmp("Elasticsearch大學(xué)");
        jobDetail.setEdu("本科及以上");
        jobDetail.setExp("一年工作經(jīng)驗(yàn)");
        jobDetail.setTitle("大數(shù)據(jù)工程師");
        jobDetail.setJob_type("全職");
        jobDetail.setPv("1700次瀏覽");
        jobDetail.setJd("會(huì)Hadoop就行");
        jobDetail.setSalary("5-9千/月");

        jobFullTextService.add(jobDetail);
    }

    @Test
    public void getTest() throws IOException {
        System.out.println(jobFullTextService.findById(1));
    }

    @Test
    public void updateTest() throws IOException {
        JobDetail jobDetail = jobFullTextService.findById(1);
        jobDetail.setTitle("大數(shù)據(jù)巨牛開(kāi)發(fā)工程師");
        jobDetail.setSalary("10W飄20W/月");

        jobFullTextService.update(jobDetail);
        System.out.println(jobFullTextService.findById(1));
    }

    @Test
    public void deleteTest() throws IOException {
        jobFullTextService.deleteById(1);
        // 執(zhí)行下句會(huì)出現(xiàn)空指針異常
        System.out.println(jobFullTextService.findById(1));
    }

    @Test
    public void searchTest() throws IOException {
        List<JobDetail> jobDetailList = jobFullTextService.searchByKeywords("flink");
        for (JobDetail jobDetail : jobDetailList) {
            System.out.println(jobDetail);
        }
    }

    @Test
    public void searchByPageTest() throws IOException {
        Map<String, Object> resultMap = jobFullTextService.searchByPage("hbase", 1, 3);
        System.out.println("一共查詢到:" + resultMap.get("total").toString());

        ArrayList<JobDetail> content = (ArrayList<JobDetail>) resultMap.get("content");
        for (JobDetail jobDetail : content) {
            System.out.println(jobDetail);
        }
    }

    @Test
    public void searchByScrollPageTest1() throws IOException {
        Map<String, Object> resultMap = jobFullTextService.searchByScrollPage("spark", null, 10);
        //scroll_id:DXF1ZXJ5QW5kRmV0Y2gBAAAAAAAAABgWTnFna0I1V2ZRbkNHTXlSbl9YTTF2UQ==
        System.out.println("scroll_id:" + resultMap.get("scroll_id").toString());

        ArrayList<JobDetail> content = (ArrayList<JobDetail>) resultMap.get("content");
        for (JobDetail jobDetail : content) {
            System.out.println(jobDetail);
        }
    }

    @Test
    public void searchByScrollPageTest2() throws IOException {
        Map<String, Object> resultMap = jobFullTextService.searchByScrollPage("spark", "DXF1ZXJ5QW5kRmV0Y2gBAAAAAAAAABgWTnFna0I1V2ZRbkNHTXlSbl9YTTF2UQ==", 10);
        System.out.println("scroll_id:" + resultMap.get("scroll_id").toString());

        ArrayList<JobDetail> content = (ArrayList<JobDetail>)resultMap.get("content");
        for (JobDetail jobDetail : content) {
            System.out.println(jobDetail);
        }
    }
    
    @After
    public void afterTest() throws IOException {
        jobFullTextService.close();
    }
}

二、Elasticsearch SQL

Elasticsearch SQL允許執(zhí)行類SQL的查詢，可以使用REST接口、命令行或者是JDBC，都可以使用SQL來(lái)進(jìn)行數(shù)據(jù)的檢索和數(shù)據(jù)的聚合。
Elasticsearch SQL特點(diǎn)：
本地集成：Elasticsearch SQL是專門為Elasticsearch構(gòu)建的。每個(gè)SQL查詢都根據(jù)底層存儲(chǔ)對(duì)相關(guān)節(jié)點(diǎn)有效執(zhí)行。
沒(méi)有額外的要求： 不依賴其他的硬件、進(jìn)程、運(yùn)行時(shí)庫(kù)，Elasticsearch SQL可以直接運(yùn)行在Elasticsearch集群上
輕量且高效：像SQL那樣簡(jiǎn)潔、高效地完成查詢

1、SQL與Elasticsearch對(duì)應(yīng)關(guān)系

2、Elasticsearch SQL語(yǔ)法

目前FROM只支持單表

SELECT select_expr [, ...]
[ FROM table_name ]
[ WHERE condition ]
[ GROUP BY grouping_element [, ...] ]
[ HAVING condition]
[ ORDER BY expression [ ASC | DESC ] [, ...] ]
[ LIMIT [ count ] ][
PIVOT ( aggregation_expr FOR column IN ( value [ [ AS ] alias ] [, ...] ) ) ]

3、示例

1）、查詢職位索引庫(kù)中的一條數(shù)據(jù)

format：表示指定返回的數(shù)據(jù)類型

// 1. 查詢職位信息
GET /_sql?format=txt
{
    "query": "SELECT * FROM job_idx limit 1"
}

除了txt類型，Elasticsearch SQL還支持以下類型

// 1. 查詢職位信息
GET /_sql?format=json
{
    "query": "SELECT * FROM job_idx limit 10"
}

2）、將SQL轉(zhuǎn)換為DSL

GET /_sql/translate
{
    "query": "SELECT * FROM job_idx limit 1"
}

查詢結(jié)果如下：

3）、職位scroll分頁(yè)查詢

第一次查詢

// 2. scroll分頁(yè)查詢
GET /_sql?format=json
{
    "query": "SELECT * FROM job_idx",
    "fetch_size": 10
}
fetch_size表示每頁(yè)顯示多少數(shù)據(jù)，而且當(dāng)我們指定format為Json格式時(shí)，會(huì)返回一個(gè)cursor ID
{
    "columns": [
        {            "name": "area",            "type": "text"        },
        {            "name": "cmp",            "type": "text"        },
        {            "name": "edu",            "type": "keyword"        },
        {            "name": "exp",            "type": "text"        },
        {            "name": "jd",            "type": "text"        },
        {            "name": "job_type",            "type": "keyword"        },
        {            "name": "pv",            "type": "keyword"        },
        {            "name": "salary",            "type": "keyword"        },
        {            "name": "title",            "type": "text"        }
    ],
    "rows": [
        [
            "工作地區(qū)：上海",
            "東普科技",
            "本科以上",
            "3年以上",
            "工作描述：1. 接口設(shè)計(jì)與開(kāi)發(fā)，能夠使用緩存與隊(duì)列，解決接口訪問(wèn)并發(fā)問(wèn)題2. 應(yīng)用系統(tǒng)設(shè)計(jì)與開(kāi)發(fā)，查詢報(bào)表制作。3. PHP程序設(shè)計(jì)與開(kāi)發(fā)4. 功能模塊開(kāi)發(fā)，代碼編寫(xiě)5. 部分需求分析與開(kāi)發(fā)文檔撰寫(xiě)6. 與代碼質(zhì)量保證與測(cè)試7. 與測(cè)試部門緊密配合，修改BUG任職要求：1. 統(tǒng)招本科以上學(xué)歷，三年以上工作經(jīng)驗(yàn)，熟悉php+mysql開(kāi)發(fā)或java+db，能夠獨(dú)立分析設(shè)計(jì)系統(tǒng)，了解至少一種js語(yǔ)言庫(kù)，例如jquery。2. 熟悉一種隊(duì)列處理機(jī)制，能夠用隊(duì)列做應(yīng)用。3. 能夠分析并優(yōu)化慢查詢sql，數(shù)據(jù)庫(kù)簡(jiǎn)單管理",
            "全職",
            "2624人瀏覽過(guò)",
            "￥ 13.2-19.2萬(wàn)元",
            "PHP開(kāi)發(fā)工程師"
        ],
        [
            。。。
        ],
        [
            。。。
        ],
        [
            。。。
        ],
        [
            。。。
        ]
    ],
    "cursor": "5/WuAwFaAXNARFhGMVpYSjVRVzVrUm1WMFkyZ0JBQUFBQUFBQUFCMFdUbkZuYTBJMVYyWlJia05IVFhsU2JsOVlUVEYyVVE9Pf8PCQFmBGFyZWEBBGFyZWEBBHRleHQAAAABZgNjbXABA2NtcAEEdGV4dAAAAAFmA2VkdQEDZWR1AQdrZXl3b3JkAQAAAWYDZXhwAQNleHABBHRleHQAAAABZgJqZAECamQBBHRleHQAAAABZghqb2JfdHlwZQEIam9iX3R5cGUBB2tleXdvcmQBAAABZgJwdgECcHYBB2tleXdvcmQBAAABZgZzYWxhcnkBBnNhbGFyeQEHa2V5d29yZAEAAAFmBXRpdGxlAQV0aXRsZQEEdGV4dAAAAAL/AQ=="
}

默認(rèn)快照的失效時(shí)間為45s，如果要延遲快照失效時(shí)間，可以配置為以下：

GET /_sql?format=json
{
    "query": "select * from job_idx",
    "fetch_size": 1000,
    "page_timeout": "10m"
}

第二次查詢

GET /_sql?format=json
{
        "cursor": "5/WuAwFaAXNARFhGMVpYSjVRVzVrUm1WMFkyZ0JBQUFBQUFBQUFCMFdUbkZuYTBJMVYyWlJia05IVFhsU2JsOVlUVEYyVVE9Pf8PCQFmBGFyZWEBBGFyZWEBBHRleHQAAAABZgNjbXABA2NtcAEEdGV4dAAAAAFmA2VkdQEDZWR1AQdrZXl3b3JkAQAAAWYDZXhwAQNleHABBHRleHQAAAABZgJqZAECamQBBHRleHQAAAABZghqb2JfdHlwZQEIam9iX3R5cGUBB2tleXdvcmQBAAABZgJwdgECcHYBB2tleXdvcmQBAAABZgZzYWxhcnkBBnNhbGFyeQEHa2V5d29yZAEAAAFmBXRpdGxlAQV0aXRsZQEEdGV4dAAAAAL/AQ=="
}

4)、清除游標(biāo)

POST /_sql/close
{
"cursor": "5/WuAwFaAXNARFhGMVpYSjVRVzVrUm1WMFkyZ0JBQUFBQUFBQUFCNFdUbkZuYTBJMVYyWlJia05IVFhsU2JsOVlUVEYyVVE9Pf8PCQFmBGFyZWEBBGFyZWEBBHRleHQAAAABZgNjbXABA2NtcAEEdGV4dAAAAAFmA2VkdQEDZWR1AQdrZXl3b3JkAQAAAWYDZXhwAQNleHABBHRleHQAAAABZgJqZAECamQBBHRleHQAAAABZghqb2JfdHlwZQEIam9iX3R5cGUBB2tleXdvcmQBAAABZgJwdgECcHYBB2tleXdvcmQBAAABZgZzYWxhcnkBBnNhbGFyeQEHa2V5d29yZAEAAAFmBXRpdGxlAQV0aXRsZQEEdGV4dAAAAAL/AQ=="
}

5）、職位全文檢索

檢索title和jd中包含hadoop的職位
在執(zhí)行全文檢索時(shí)，需要使用到MATCH函數(shù)
MATCH(field_exp,constant_exp[, options])
field_exp：匹配字段
constant_exp：匹配常量表達(dá)式

GET /_sql?format=txt
{
    "query": "select * from job_idx where MATCH(title, 'hadoop') or MATCH(jd, 'hadoop') limit 10"
 }

6）、訂單統(tǒng)計(jì)分析案例

基于按數(shù)據(jù)，使用Elasticsearch中的聚合統(tǒng)計(jì)功能

1、創(chuàng)建索引

PUT /order_idx/
{
    "mappings": {
        "properties": {
            "id": {
                "type": "keyword",
                "store": true
            },
            "status": {
                "type": "keyword",
                "store": true
            },
            "pay_money": {
                "type": "double",
                "store": true
            },
            "payway": {
                "type": "byte",
                "store": true
            },
            "userid": {
                "type": "keyword",
                "store": true
            },
            "operation_date": {
                "type": "date",
                "format": "yyyy-MM-dd HH:mm:ss",
                "store": true
            },
            "category": {
                "type": "keyword",
                "store": true
            }
        }
    }
}

2、導(dǎo)入測(cè)試數(shù)據(jù)

curl -H "Content-Type: application/json" -XPOST "server1:9200/order_idx/_bulk?pretty&refresh" --data-binary "@order_data.json"

3、統(tǒng)計(jì)不同支付方式的的訂單數(shù)量

• 使用JSON DSL的方式來(lái)實(shí)現(xiàn)
  用Elasticsearch原生支持的基于JSON的DSL方式來(lái)實(shí)現(xiàn)聚合統(tǒng)計(jì)

GET /order_idx/_search
{
    "size": 0,
    "aggs": {
        "group_by_state": {
            "terms": {
                "field": "payway"
            }
        }
    }
}

這種方式分析起來(lái)比較麻煩。ES也能夠使用SQL方式來(lái)進(jìn)行統(tǒng)計(jì)和分析的

• 基于Elasticsearch SQL方式實(shí)現(xiàn)

ET /_sql?format=txt
{
    "query": "select payway, count(*) as order_cnt from order_idx group by payway"
}

• 基于JDBC方式統(tǒng)計(jì)不同方式的訂單數(shù)量
  pom.xml
  該版本在maven上沒(méi)有，需要使用阿里的鏡像庫(kù)

<dependency>
            <groupId>org.elasticsearch.plugin</groupId>
            <artifactId>x-pack-sql-jdbc</artifactId>
            <version>7.6.1</version>
</dependency>

開(kāi)啟X-pack高階功能試用，如果不開(kāi)啟試用，會(huì)報(bào)如下錯(cuò)誤
current license is non-compliant for [jdbc]

在server1節(jié)點(diǎn)上執(zhí)行：
查看服務(wù)器es的license信息，發(fā)現(xiàn) “type” : “basic”

# curl -XGET http://server1:9200/_license
{
  "license" : {
    "status" : "active",
    "uid" : "91546f48-bd7f-4a74-b4b9-889dece7db80",
    "type" : "basic",
    "issue_date" : "2020-05-12T20:10:42.742Z",
    "issue_date_in_millis" : 1589314242742,
    "max_nodes" : 1000,
    "issued_to" : "my-application",
    "issuer" : "elasticsearch",
    "start_date_in_millis" : -1
  }
}


#修改成30天試用版，https://www.elastic.co/guide/en/elasticsearch/reference/master/start-trial.html


curl http://server1:9200/_license/start_trial?acknowledge=true  -X POST {"acknowledged":true,"trial_was_started":true,"type":"trial"}
試用期為30天。

[alanchan@server1 testdata]$ curl http://server1:9200/_license/start_trial?acknowledge=true  -X POST {"acknowledged":true,"trial_was_started":true,"type":"trial"}
curl: (6) Couldn't resolve host 'acknowledged:true'
curl: (6) Couldn't resolve host 'trial_was_started:true'
curl: (6) Couldn't resolve host 'type:trial'
{"acknowledged":true,"trial_was_started":true,"type":"trial"}
[alanchan@server1 testdata]$ 

#再查看license信息時(shí)，內(nèi)容已發(fā)生變化
# curl -XGET http://localhost:9200/_license
{
  "license" : {
    "status" : "active",
    "uid" : "275f23b1-7b57-4bfd-b309-16d9545aebfa",
    "type" : "trial",
    "issue_date" : "2020-05-14T03:16:54.139Z",
    "issue_date_in_millis" : 1589426214139,
    "expiry_date" : "2020-06-13T03:16:54.139Z",
    "expiry_date_in_millis" : 1592018214139,
    "max_nodes" : 1000,
    "issued_to" : "my-application",
    "issuer" : "elasticsearch",
    "start_date_in_millis" : -1
  }
}


[alanchan@server1 testdata]$ 

• 源碼

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

import org.elasticsearch.xpack.sql.jdbc.EsDriver;

public class JdbcExample {

    public static void main(String[] args) throws Exception {
        // 1. 加載ES驅(qū)動(dòng)
        Class.forName(EsDriver.class.getName());

        // 2. 建立連接
        Connection connection = DriverManager.getConnection("jdbc:es://http://server1:9200");

        // 3. 準(zhǔn)備SQL語(yǔ)句
        String sql = "select payway, count(*) cnt from order_idx group by payway";

        // 4. 使用PreparedStatement執(zhí)行SQL
        PreparedStatement preparedStatement = connection.prepareStatement(sql);
        ResultSet resultSet = preparedStatement.executeQuery();

        // 5. 遍歷結(jié)果
        while (resultSet.next()) {
            byte payway = resultSet.getByte("payway");
            long cnt = resultSet.getLong("cnt");

            System.out.println("支付方式：" + payway + " 訂單數(shù)量：" + cnt);
        }

        // 6. 關(guān)閉連接
        resultSet.close();
        connection.close();
    }

}

• 運(yùn)行結(jié)果
  

4、統(tǒng)計(jì)不同支付方式訂單數(shù)，并按照訂單數(shù)量倒序排序

GET /_sql?format=txt
{
    "query": "select payway, count(*) as order_cnt from order_idx group by payway order by order_cnt desc"
}

5、只統(tǒng)計(jì)「已付款」?fàn)顟B(tài)的不同支付方式的訂單數(shù)量

GET /_sql?format=txt
{
    "query": "select payway, count(*) as order_cnt from order_idx where status = '已付款' group by payway order by order_cnt desc"
}

6、統(tǒng)計(jì)不同用戶的總訂單數(shù)量、總訂單金額

GET /_sql?format=txt
{
    "query": "select userid, count(1) as cnt, sum(pay_money) as total_money from order_idx group by userid"
}

4、Elasticsearch SQL目前的一些限制

目前Elasticsearch SQL還存在一些限制。
例如：不支持JOIN、不支持較復(fù)雜的子查詢。所以，有一些相對(duì)復(fù)雜一些的功能，還得借助于DSL方式來(lái)實(shí)現(xiàn)

以上，簡(jiǎn)單的介紹了java api操作Elasticsearch和Elasticsearch SQL的詳細(xì)示例。文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-575393.html

到了這里，關(guān)于4、Elasticsearch7.6.1 Java api操作ES（CRUD、兩種分頁(yè)方式、高亮顯示）和Elasticsearch SQL詳細(xì)示例的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！