ElasticSearch的基本原理與用法

時間  2019-11-13
標籤 elasticsearch 基本原理 用法 欄目 日誌分析 简体版
原文   原文鏈接

1、簡介html

ElasticSearch和Solr都是基於Lucene的搜索引擎，不過ElasticSearch天生支持分佈式，而Solr是4.0版本後的SolrCloud纔是分佈式版本，Solr的分佈式支持須要ZooKeeper的支持。java

這裏有一個詳細的ElasticSearch和Solr的對比：http://solr-vs-elasticsearch.com/node

2、基本用法git

集羣（Cluster）： ES是一個分佈式的搜索引擎，通常由多臺物理機組成。這些物理機，經過配置一個相同的cluster name，互相發現，把本身組織成一個集羣。github

節點（Node)：同一個集羣中的一個Elasticsearch主機。sql

Node類型：數據庫

1）data node: 存儲index數據。Data nodes hold data and perform data related operations such as CRUD, search, and aggregations.json

2）client node: 不存儲index，處理轉發客戶端請求到Data Node。api

3）master node: 不存儲index，集羣管理，如管理路由信息（routing infomation），判斷node是否available，當有node出現或消失時重定位分片（shards），當有node failure時協調恢復。（全部的master node會選舉出一個master leader node）緩存

詳情參考：https://www.elastic.co/guide/en/elasticsearch/reference/current/modules-node.html

主分片（Primary shard）：索引（下文介紹）的一個物理子集。同一個索引在物理上能夠切多個分片，分佈到不一樣的節點上。分片的實現是Lucene 中的索引。

注意：ES中一個索引的分片個數是創建索引時就要指定的，創建後不可再改變。因此開始建一個索引時，就要預計數據規模，將分片的個數分配在一個合理的範圍。

副本分片（Replica shard）：每一個主分片能夠有一個或者多個副本，個數是用戶本身配置的。ES會盡可能將同一索引的不一樣分片分佈到不一樣的節點上，提升容錯性。對一個索引，只要不是全部shards所在的機器都掛了，就還能用。

索引（Index)：邏輯概念，一個可檢索的文檔對象的集合。相似與DB中的database概念。同一個集羣中可創建多個索引。好比，生產環境常見的一種方法，對每月產生的數據建索引，以保證單個索引的量級可控。

類型（Type）：索引的下一級概念，大概至關於數據庫中的table。同一個索引裏能夠包含多個 Type。 

文檔（Document)：即搜索引擎中的文檔概念，也是ES中一個能夠被檢索的基本單位，至關於數據庫中的row，一條記錄。

字段（Field）：至關於數據庫中的column。ES中，每一個文檔，實際上是以json形式存儲的。而一個文檔能夠被視爲多個字段的集合。好比一篇文章，可能包括了主題、摘要、正文、做者、時間等信息，每一個信息都是一個字段，最後被整合成一個json串，落地到磁盤。

映射（Mapping）：至關於數據庫中的schema，用來約束字段的類型，不過 Elasticsearch 的 mapping 能夠不顯示地指定、自動根據文檔數據建立。

Elasticsearch集羣能夠包含多個索引（indices），每個索引能夠包含多個類型（types），每個類型包含多個文檔（documents），而後每一個文檔包含多個字段（Fields），這種面向文檔型的儲存，也算是NoSQL的一種吧。

ES比傳統關係型數據庫，對一些概念上的理解：

Relational DB -> Databases -> Tables -> Rows -> Columns
Elasticsearch -> Indices   -> Types  -> Documents -> Fields

從建立一個Client到添加、刪除、查詢等基本用法：

一、建立Client

public ElasticSearchService(String ipAddress, int port) {
        client = new TransportClient()
                .addTransportAddress(new InetSocketTransportAddress(ipAddress,
                        port));
    }

這裏是一個TransportClient。

ES下兩種客戶端對比：

TransportClient：輕量級的Client，使用Netty線程池，Socket鏈接到ES集羣。自己不加入到集羣，只做爲請求的處理。

Node Client：客戶端節點自己也是ES節點，加入到集羣，和其餘ElasticSearch節點同樣。頻繁的開啓和關閉這類Node Clients會在集羣中產生「噪音」。

二、建立/刪除Index和Type信息

    // 建立索引
    public void createIndex() {
        client.admin().indices().create(new CreateIndexRequest(IndexName))
                .actionGet();
    }

    // 清除全部索引
    public void deleteIndex() {
        IndicesExistsResponse indicesExistsResponse = client.admin().indices()
                .exists(new IndicesExistsRequest(new String[] { IndexName }))
                .actionGet();
        if (indicesExistsResponse.isExists()) {
            client.admin().indices().delete(new DeleteIndexRequest(IndexName))
                    .actionGet();
        }
    }
    
    // 刪除Index下的某個Type
    public void deleteType(){        client.prepareDelete().setIndex(IndexName).setType(TypeName).execute().actionGet();
    }

    // 定義索引的映射類型
    public void defineIndexTypeMapping() {
        try {
            XContentBuilder mapBuilder = XContentFactory.jsonBuilder();
            mapBuilder.startObject()
            .startObject(TypeName)
            .startObject("_all").field("enabled", false).endObject()
                .startObject("properties")
                    .startObject(IDFieldName).field("type", "long").endObject()
                    .startObject(SeqNumFieldName).field("type", "long").endObject()
                    .startObject(IMSIFieldName).field("type", "string").field("index", "not_analyzed").endObject()
                    .startObject(IMEIFieldName).field("type", "string").field("index", "not_analyzed").endObject()
                    .startObject(DeviceIDFieldName).field("type", "string").field("index", "not_analyzed").endObject()
                    .startObject(OwnAreaFieldName).field("type", "string").field("index", "not_analyzed").endObject()
                    .startObject(TeleOperFieldName).field("type", "string").field("index", "not_analyzed").endObject()
                    .startObject(TimeFieldName).field("type", "date").field("store", "yes").endObject()
                .endObject()
            .endObject()
            .endObject();

            PutMappingRequest putMappingRequest = Requests
                    .putMappingRequest(IndexName).type(TypeName)
                    .source(mapBuilder);
            client.admin().indices().putMapping(putMappingRequest).actionGet();
        } catch (IOException e) {
            log.error(e.toString());
        }

這裏自定義了某個Type的索引映射（Mapping）：

1）默認ES會自動處理數據類型的映射：針對整型映射爲long，浮點數爲double，字符串映射爲string，時間爲date，true或false爲boolean。

2）字段的默認配置是indexed，但不是stored的，也就是 field("index", "yes").field("store", "no")。

3）這裏Disable了「_all」字段，_all字段會把全部的字段用空格鏈接，而後用「analyzed」的方式index這個字段，這個字段能夠被search，可是不能被retrieve。

4）針對string，ES默認會作「analyzed」處理，即先作分詞、去掉stop words等處理再index。若是你須要把一個字符串作爲總體被索引到，須要把這個字段這樣設置：field("index", "not_analyzed")。

5）默認_source字段是enabled，_source字段存儲了原始Json字符串（original JSON document body that was passed at index time）。

詳情參考：

https://www.elastic.co/guide/en/elasticsearch/guide/current/mapping-intro.html

https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping-store.html

https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping-all-field.html

https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping-source-field.html

三、索引數據

    // 批量索引數據
    public void indexHotSpotDataList(List<Hotspotdata> dataList) {
        if (dataList != null) {
            int size = dataList.size();
            if (size > 0) {
                BulkRequestBuilder bulkRequest = client.prepareBulk();
                for (int i = 0; i < size; ++i) {
                    Hotspotdata data = dataList.get(i);
                    String jsonSource = getIndexDataFromHotspotData(data);
                    if (jsonSource != null) {
                        bulkRequest.add(client
                                .prepareIndex(IndexName, TypeName,
                                        data.getId().toString())
                                .setRefresh(true).setSource(jsonSource));
                    }
                }

                BulkResponse bulkResponse = bulkRequest.execute().actionGet();
                if (bulkResponse.hasFailures()) {
                    Iterator<BulkItemResponse> iter = bulkResponse.iterator();
                    while (iter.hasNext()) {
                        BulkItemResponse itemResponse = iter.next();
                        if (itemResponse.isFailed()) {
                            log.error(itemResponse.getFailureMessage());
                        }
                    }
                }
            }
        }
    }

    // 索引數據
    public boolean indexHotspotData(Hotspotdata data) {
        String jsonSource = getIndexDataFromHotspotData(data);
        if (jsonSource != null) {
            IndexRequestBuilder requestBuilder = client.prepareIndex(IndexName,
                    TypeName).setRefresh(true);
            requestBuilder.setSource(jsonSource)
                    .execute().actionGet();
            return true;
        }

        return false;
    }

    // 獲得索引字符串
    public String getIndexDataFromHotspotData(Hotspotdata data) {
        String jsonString = null;
        if (data != null) {
            try {
                XContentBuilder jsonBuilder = XContentFactory.jsonBuilder();
                jsonBuilder.startObject().field(IDFieldName, data.getId())
                        .field(SeqNumFieldName, data.getSeqNum())
                        .field(IMSIFieldName, data.getImsi())
                        .field(IMEIFieldName, data.getImei())
                        .field(DeviceIDFieldName, data.getDeviceID())
                        .field(OwnAreaFieldName, data.getOwnArea())
                        .field(TeleOperFieldName, data.getTeleOper())
                        .field(TimeFieldName, data.getCollectTime())
                        .endObject();
                jsonString = jsonBuilder.string();
            } catch (IOException e) {
                log.equals(e);
            }
        }

        return jsonString;
    }

ES支持批量和單個數據索引。

四、查詢獲取數據

    // 獲取少許數據100個
    private List<Integer> getSearchData(QueryBuilder queryBuilder) {
        List<Integer> ids = new ArrayList<>();
        SearchResponse searchResponse = client.prepareSearch(IndexName)
                .setTypes(TypeName).setQuery(queryBuilder).setSize(100)
                .execute().actionGet();
        SearchHits searchHits = searchResponse.getHits();
        for (SearchHit searchHit : searchHits) {
            Integer id = (Integer) searchHit.getSource().get("id");
            ids.add(id);
        }
        return ids;
    }

    // 獲取大量數據
    private List<Integer> getSearchDataByScrolls(QueryBuilder queryBuilder) {
        List<Integer> ids = new ArrayList<>();
        // 一次獲取100000數據
        SearchResponse scrollResp = client.prepareSearch(IndexName)
                .setSearchType(SearchType.SCAN).setScroll(new TimeValue(60000))
                .setQuery(queryBuilder).setSize(100000).execute().actionGet();
        while (true) {
            for (SearchHit searchHit : scrollResp.getHits().getHits()) {
                Integer id = (Integer) searchHit.getSource().get(IDFieldName);
                ids.add(id);
            }
            scrollResp = client.prepareSearchScroll(scrollResp.getScrollId())
                    .setScroll(new TimeValue(600000)).execute().actionGet();
            if (scrollResp.getHits().getHits().length == 0) {
                break;
            }
        }

        return ids;
    }

這裏的QueryBuilder是一個查詢條件，ES支持分頁查詢獲取數據，也能夠一次性獲取大量數據，須要使用Scroll Search。

五、聚合（Aggregation Facet）查詢 

    // 獲得某段時間內設備列表上每一個設備的數據分佈狀況<設備ID，數量>
    public Map<String, String> getDeviceDistributedInfo(String startTime,
            String endTime, List<String> deviceList) {

        Map<String, String> resultsMap = new HashMap<>();

        QueryBuilder deviceQueryBuilder = getDeviceQueryBuilder(deviceList);
        QueryBuilder rangeBuilder = getDateRangeQueryBuilder(startTime, endTime);
        QueryBuilder queryBuilder = QueryBuilders.boolQuery()
                .must(deviceQueryBuilder).must(rangeBuilder);

        TermsBuilder termsBuilder = AggregationBuilders.terms("DeviceIDAgg").size(Integer.MAX_VALUE)
                .field(DeviceIDFieldName);
        SearchResponse searchResponse = client.prepareSearch(IndexName)
                .setQuery(queryBuilder).addAggregation(termsBuilder)
                .execute().actionGet();
        Terms terms = searchResponse.getAggregations().get("DeviceIDAgg");
        if (terms != null) {
            for (Terms.Bucket entry : terms.getBuckets()) {
                resultsMap.put(entry.getKey(),
                        String.valueOf(entry.getDocCount()));
            }
        }
        return resultsMap;
    }

Aggregation查詢能夠查詢相似統計分析這樣的功能：如某個月的數據分佈狀況，某類數據的最大、最小、總和、平均值等。

詳情參考：https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/java-aggs.html

3、集羣配置

配置文件elasticsearch.yml

集羣名和節點名：

#cluster.name: elasticsearch

#node.name: "Franz Kafka"

是否參與master選舉和是否存儲數據

#node.master: true

#node.data: true

分片數和副本數

#index.number_of_shards: 5
#index.number_of_replicas: 1

容許其餘網絡訪問:  

network.host: 0

master選舉最少的節點數，這個必定要設置爲整個集羣節點個數的一半加1，即N/2+1

#discovery.zen.minimum_master_nodes: 1

discovery ping的超時時間，擁塞網絡，網絡狀態不佳的狀況下設置高一點

#discovery.zen.ping.timeout: 3s

注意，分佈式系統整個集羣節點個數N要爲奇數個！！

如何避免ElasticSearch發生腦裂（brain split）：http://blog.trifork.com/2013/10/24/how-to-avoid-the-split-brain-problem-in-elasticsearch/

即便集羣節點個數爲奇數，minimum_master_nodes爲整個集羣節點個數一半加1，也難以免腦裂的發生，詳情看討論：https://github.com/elastic/elasticsearch/issues/2488

4、經常使用查詢

curl -X<REST Verb> <Node>:<Port>/<Index>/<Type>/<ID>

Index info:

curl -XGET 'localhost:9200'
curl -XGET 'localhost:9200/_stats?pretty'
curl -XGET 'localhost:9200/{index}/_stats?pretty'
curl -XGET 'localhost:9200/_cluster/health?level=indices&pretty=true'
curl -XGET 'localhost:9200/{index}?pretty'
curl -XGET 'localhost:9200/_cat/indices?v'
curl -XGET 'localhost:9200/{index}/_mapping/{type}?pretty'

Mapping info:
curl -XGET 'localhost:9200/subscriber/_mapping/subscriber?pretty'

Index search:
curl -XGET 'localhost:9200/subscriber/subscriber/_search?pretty'

Search by ID:
curl -XGET 'localhost:9200/subscriber/subscriber/5000?pretty'

Search by field:
curl -XGET 'localhost:9200/subscriber/subscriber/_search?q=ipAddress:63.141.15.45&&pretty'

Delete index:
curl -XDELETE 'localhost:9200/subscriber?pretty'

Delete document by ID:
curl -XDELETE 'localhost:9200/subscriber/subscriber/5000?pretty'

Delete document by query:
curl -XDELETE 'localhost:9200/subscriber/subscriber/_query?q=ipAddress:63.141.15.45&&pretty'

5、基本原理

一、ES寫數據原理

每一個doc，經過以下公式決定寫到哪一個分片上：

shard= hash(routing) % number_of_primary_shards

Routing 是一個可變值，默認是文檔的 _id ，也能夠自定義一個routing規則。

默認狀況下，primary shard在寫操做前，須要肯定大多數（a quorum, or majority）的shard copies是可用的。這樣是爲了防止在有網絡分區（network partition）的狀況下把數據寫到了錯誤的分區。

A quorum是由如下公式決定：

int( (primary + number_of_replicas) / 2 ) + 1，number_of_replicas是在index settings中指定的複製個數。

肯定一致性的值有：one (只有primary shard)，all (the primary and all replicas)，或者是默認的quorum。

若是沒有足夠可用的shard copies，elasticsearch會等待直到超時，默認等待一分鐘。

  • 一個新文檔被索引以後，先被寫入到內存中，可是爲了防止數據的丟失，會追加一份數據到事務日誌（trans log）中。 不斷有新的文檔被寫入到內存，同時也都會記錄到事務日誌中。這時新數據還不能被檢索和查詢。
  • 當達到默認的刷新時間或內存中的數據達到必定量後，會觸發一次 Refresh，將內存中的數據以一個新段形式刷新到文件緩存系統中並清空內存。這時雖然新段未被提交到磁盤，可是能夠提供文檔的檢索功能且不能被修改。
  • 隨着新文檔索引不斷被寫入，當日志數據大小超過 512M 或者時間超過 30 分鐘時，會觸發一次 Flush。 內存中的數據被寫入到一個新段同時被寫入到文件緩存系統，文件系統緩存中數據經過 Fsync 刷新到磁盤中，生成提交點，日誌文件被刪除，建立一個空的新日誌。

二、ES讀數據原理

Elasticsearch中的查詢主要分爲兩類，Get請求：經過ID查詢特定Doc；Search請求：經過Query查詢匹配Doc。

  • 對於Search類請求，查詢的時候是一塊兒查詢內存和磁盤上的Segment，最後將結果合併後返回。這種查詢是近實時（Near Real Time）的，主要是因爲內存中的Index數據須要一段時間後纔會刷新爲Segment。
  • 對於Get類請求，查詢的時候是先查詢內存中的TransLog，若是找到就當即返回，若是沒找到再查詢磁盤上的TransLog，若是尚未則再去查詢磁盤上的Segment。這種查詢是實時（Real Time）的。這種查詢順序能夠保證查詢到的Doc是最新版本的Doc，這個功能也是爲了保證NoSQL場景下的實時性要求。

全部的搜索系統通常都是兩階段查詢，第一階段查詢到匹配的DocID，第二階段再查詢DocID對應的完整文檔，這種在Elasticsearch中稱爲query_then_fetch，還有一種是一階段查詢的時候就返回完整Doc，在Elasticsearch中稱做query_and_fetch，通常第二種適用於只須要查詢一個Shard的請求。

除了一階段，兩階段外，還有一種三階段查詢的狀況。搜索裏面有一種算分邏輯是根據TF（Term Frequency）和DF（Document Frequency）計算基礎分，可是Elasticsearch中查詢的時候，是在每一個Shard中獨立查詢的，每一個Shard中的TF和DF也是獨立的，雖然在寫入的時候經過_routing保證Doc分佈均勻，可是無法保證TF和DF均勻，那麼就有會致使局部的TF和DF不許的狀況出現，這個時候基於TF、DF的算分就不許。爲了解決這個問題，Elasticsearch中引入了DFS查詢，好比DFS_query_then_fetch，會先收集全部Shard中的TF和DF值，而後將這些值帶入請求中，再次執行query_then_fetch，這樣算分的時候TF和DF就是準確的，相似的有DFS_query_and_fetch。這種查詢的優點是算分更加精準，可是效率會變差。另外一種選擇是用BM25代替TF/DF模型。

在新版本Elasticsearch中，用戶無法指定DFS_query_and_fetch和query_and_fetch，這兩種只能被Elasticsearch系統改寫。

6、Elasticsearch插件

一、elasticsearch-head是一個elasticsearch的集羣管理工具：./elasticsearch-1.7.1/bin/plugin -install mobz/elasticsearch-head

github地址：https://github.com/mobz/elasticsearch-head

二、elasticsearch-sql：使用SQL語法查詢elasticsearch：./bin/plugin -u https://github.com/NLPchina/elasticsearch-sql/releases/download/1.3.5/elasticsearch-sql-1.3.5.zip --install sql

github地址：https://github.com/NLPchina/elasticsearch-sql

三、elasticsearch-bigdesk是elasticsearch的一個集羣監控工具，能夠經過它來查看ES集羣的各類狀態。

安裝：./bin/plugin -install lukas-vlcek/bigdesk

訪問：http://192.103.101.203:9200/_plugin/bigdesk/

github地址：https://github.com/hlstudio/bigdesk

四、elasticsearch-servicewrapper插件是ElasticSearch的服務化插件

https://github.com/elasticsearch/elasticsearch-servicewrapper 

DEPRECATED: The service wrapper is deprecated and not maintained. 該項目已再也不維護。 

 

例子代碼在GitHub上：https://github.com/luxiaoxun/Code4Java

 

參考：

https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/index.html

https://www.elastic.co/guide/en/elasticsearch/reference/current/index.html

https://www.elastic.co/guide/en/elasticsearch/guide/current/distrib-write.html

http://stackoverflow.com/questions/10213009/solr-vs-elasticsearch

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程