es搜索引擎

一、es簡介

二、es優缺點

三、es使用

四、es能夠解決的問題

五、es舉例

六、es執行結果截圖

七、es數據增量方案

八、使用es搜索

　　1、es簡介

　　　　es是一個是一個實時的分佈式搜索和分析引擎。它能夠幫助你用史無前例的速度去處理大規模數據。

　　　　它能夠用於全文搜索，結構化搜索以及分析，固然你也能夠將這三者進行組合。

　　　　es是一個創建在全文搜索引擎 Apache Lucene™ 基礎上的搜索引擎，能夠說Lucene是當今最早進，最高效的全功能開源搜索引擎框架。

 

　　　　es使用Lucene做爲內部引擎，可是在使用它作全文搜索時，只須要使用統一開發好的API便可，而不須要了解其背後複雜的Lucene的運行原理。　　　　　　

　　　　es除了作全索引外，還能夠作以下工做：

　　　　分佈式實時文件存儲，並將每個字段都編入索引，使其能夠被搜索。

　　　　實時分析的分佈式搜索引擎。

　　　　能夠擴展到上百臺服務器，處理PB級別的結構化或非結構化數據。

　　　　以上功能能夠經過你喜歡的編程語言或客戶端與es的restful api進行通信。

 

　　2、概念說了半天了，說下優缺點吧

　　　　優勢：

　　　　一、es是分佈式的，不須要其它組件，分發是實時由es主節點內部自動完成的。

　　　　二、處理多組用戶，而不需特殊配置。

　　　　三、es擦用gateway的概念，（gateway：網關是網絡鏈接設備的重要組成部分，它不只具備路由的功能，並且能在兩個不一樣的協議集之間進行轉換，從而使不一樣的網絡之間進行互聯。例如：一個Netware局域網經過網關能夠訪問IBM的SNA網絡，這樣使用IPX協議的PC就可和SNA網絡上的IBM主機進行通訊。）是得備份更簡單。

　　　　四、es節點發生故障時，能夠進行自動分配其它節點替代。

　　　　缺點：

　　　　一、文檔太少，不易維護。

　　　　二、目前以爲，建索引的速度不夠快，期待有更好的方法。

 

　　3、es使用（包括建立和搜索以及關閉）　　　　　　　　

es的獲取和關閉方法：

package com.elasticsearch.config;

import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;

import java.util.ArrayList;
import java.util.List;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.xcontent.XContentBuilder;

/**
* 初始化鏈接es服務端,這裏至關於dao層..
* 
* @author:jackkang
* 
* 2013-1-12 下午11:27:37
*/
public class InitES {

static Log log = LogFactory.getLog(InitES.class);
/**
* 靜態,單例...
*/
private static TransportClient client;

public static TransportClient initESClient() {
try {
if (client == null) {
// 配置你的es,如今這裏只配置了集羣的名,默認是elasticsearch,跟服務器的相同
Settings settings = ImmutableSettings
.settingsBuilder()
.put("cluster.name", "elasticsearch")
.put("discovery.type", "zen")//發現集羣方式
.put("discovery.zen.minimum_master_nodes", 2)//最少有2個master存在
.put("discovery.zen.ping_timeout", "200ms")//集羣ping時間，過小可能會由於網絡通訊而致使不能發現集羣
.put("discovery.initial_state_timeout", "500ms")
.put("gateway.type", "local")//(fs, none, local)
.put("index.number_of_shards", 1)
.put("action.auto_create_index", false)//配置是否自動建立索引
.put("cluster.routing.schedule", "50ms")//發現新節點時間 

.build();
// 從屬性文件中獲取搜索服務器相對域地址
String transportAddresses = Config.getProperty(
"transportAddresses", "");
// 集羣地址配置
List<InetSocketTransportAddress> list = new ArrayList<InetSocketTransportAddress>();
if (StringUtils.isNotEmpty(transportAddresses)) {
String[] strArr = transportAddresses.split(",");
for (String str : strArr) {
String[] addressAndPort = str.split(":");
String address = addressAndPort[0];
int port = Integer.valueOf(addressAndPort[1]);

InetSocketTransportAddress inetSocketTransportAddress = new InetSocketTransportAddress(
address, port);
list.add(inetSocketTransportAddress);
}
}
// 這裏能夠同時鏈接集羣的服務器,能夠多個,而且鏈接服務是可訪問的
InetSocketTransportAddress addressList[] = (InetSocketTransportAddress[]) list
.toArray(new InetSocketTransportAddress[list.size()]);
// Object addressList[]=(Object [])list.toArray();

client = new TransportClient(settings)
.addTransportAddresses(addressList);
// 這裏能夠同時鏈接集羣的服務器,能夠多個,而且鏈接服務是可訪問的 192.168.1.102
// client = new TransportClient(settings).addTransportAddresses(
// new InetSocketTransportAddress("192.168.1.103", 9300));
//
// Client client = new TransportClient()
// .addTransportAddress(new
// InetSocketTransportAddress("192.168.0.149", 9300))
// .addTransportAddress(new
// InetSocketTransportAddress("192.168.0.162", 9300));

// 改變shards數目：
/*client.admin().indices().prepareUpdateSettings("test")
.setSettings(ImmutableSettings.settingsBuilder().put("index.number_of_replicas", 2)).execute().actionGet();*/
}
} catch (Exception e) {
// if (log.isDebugEnabled()) {
// log.debug("方法AppCommentAction-deleteAppComment,參數信息：commentid" );
// }
log.error("獲取客戶端對象異常：" + e.getMessage());
}
return client;
}


public static void closeESClient() {
if (client != null) {
client.close();
}
}
}

 

 

 

搜索：

package com.elasticsearch.action;

import java.io.UnsupportedEncodingException;
import java.net.MalformedURLException;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;

import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;

import com.elasticsearch.config.ElasticsearchUtil;
import com.elasticsearch.pojo.Pager;
import com.opensymphony.xwork2.ActionSupport;

public class SearchAction extends ActionSupport {
private static final long serialVersionUID = 1L;
/** 關鍵字 **/
private String wd;
/** 消耗時間 **/
private double spendTime;
/** 查詢結果集對象 **/
private List<Map<String, Object>> pageList = new ArrayList<Map<String, Object>>();
/** 分頁對象 **/
private Pager pager;
/** 總記錄數 使用靜態變量的方式緩存 **/
private Long total;

private SearchResponse response;

/**
* 條件檢索action
* 
* @throws MalformedURLException
* @throws SolrServerException
* @throws UnsupportedEncodingException
**/
public String search() throws MalformedURLException,
UnsupportedEncodingException {
/** 檢索開始時間 **/
long startTime = System.currentTimeMillis();

/** 獲取頁面封裝好的分頁對象 **/
if (pager == null) {
pager = new Pager();
pager.setMaxPageItems(10);
}

wd = new String(wd.getBytes("ISO-8859-1"), "UTF-8"); // 解決亂碼

pager.setDefaultMaxPageItems(1);
/**高亮字段**/
String[] highFields=new String[]{"content","title"};

response = ElasticsearchUtil.searcher("medcl", "news",
pager.getOffset(), pager.getMaxPageItems(), wd,highFields);

/** 總記錄數 **/
total = response.getHits().totalHits();

System.out.println("命中總數：" + total);
SearchHits searchHits = response.getHits();
SearchHit[] hits = searchHits.getHits();
for (int i = 0; i < hits.length; i++) {
Map<String, Object> map = new HashMap<String, Object>();
SearchHit hit = hits[i];
String id=hit.getId();

String content = ElasticsearchUtil.getHighlightFields(hit,"content");
String title = ElasticsearchUtil.getHighlightFields(hit,"title");

map.put("id", hit.getSource().get("id"));
map.put("content", content);
map.put("title", title);
map.put("create_time", hit.getSource().get("create_time"));
map.put("links", hit.getSource().get("link"));

pageList.add(map);
}

/** 檢索完成時間 **/
long endTime = System.currentTimeMillis();
/** 檢索花費時間 **/
//spendTime = (double) (endTime - startTime) / 1000;


Calendar c = Calendar.getInstance(); 
c.setTimeInMillis(endTime - startTime); 
spendTime = c.get(Calendar.MILLISECOND);

return SUCCESS;
}

public static String Html2Text(String inputString) {
String htmlStr = inputString; // 含html標籤的字符串
String textStr = "";
java.util.regex.Pattern p_script;
java.util.regex.Matcher m_script;
java.util.regex.Pattern p_style;
java.util.regex.Matcher m_style;
java.util.regex.Pattern p_html;
java.util.regex.Matcher m_html;

try {
String regEx_script = "<[\\s]*?script[^>]*?>[\\s\\S]*?<[\\s]*?\\/[\\s]*?script[\\s]*?>"; // 定義script的正則表達式{或<script[^>]*?>[\\s\\S]*?<\\/script>
// }
String regEx_style = "<[\\s]*?style[^>]*?>[\\s\\S]*?<[\\s]*?\\/[\\s]*?style[\\s]*?>"; // 定義style的正則表達式{或<style[^>]*?>[\\s\\S]*?<\\/style>
// }
String regEx_html = "<[^>]+>"; // 定義HTML標籤的正則表達式

p_script = Pattern.compile(regEx_script, Pattern.CASE_INSENSITIVE);
m_script = p_script.matcher(htmlStr);
htmlStr = m_script.replaceAll(""); // 過濾script標籤

p_style = Pattern.compile(regEx_style, Pattern.CASE_INSENSITIVE);
m_style = p_style.matcher(htmlStr);
htmlStr = m_style.replaceAll(""); // 過濾style標籤

p_html = Pattern.compile(regEx_html, Pattern.CASE_INSENSITIVE);
m_html = p_html.matcher(htmlStr);
htmlStr = m_html.replaceAll(""); // 過濾html標籤

textStr = htmlStr;

} catch (Exception e) {
System.err.println("Html2Text: " + e.getMessage());
}

return textStr;// 返回文本字符串
}

public String getWd() {
return wd;
}

public void setWd(String wd) {
this.wd = wd;
}

public double getSpendTime() {
return spendTime;
}

public void setSpendTime(double spendTime) {
this.spendTime = spendTime;
}

public List<Map<String, Object>> getPageList() {
return pageList;
}

public void setPageList(List<Map<String, Object>> pageList) {
this.pageList = pageList;
}

public Pager getPager() {
return pager;
}

public void setPager(Pager pager) {
this.pager = pager;
}

public Long getTotal() {
return total;
}

public void setTotal(Long total) {
this.total = total;
}
}

 
　　4、能夠決絕基金的問題

　　　　隨着基金系統的逐漸完善，數據量的增長，使用es能夠緩解檢索數據給數據庫帶來的壓力。

　　　　好比，基金中報銷的病例列表，報銷記錄，消費記錄等

　　5、es舉例

　　　　es使用java連接，建立mapping，保存數據

　　　　demo的javabean

package com.elasticsearch;

import com.google.common.collect.Lists;
import com.util.date.Joda_Time;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;

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

/**
 * javabean
 */
public class User {

    private String name;
    private String home;//家鄉
    private double height;//身高
    private int age;
    private Date birthday;

    public User() {
    }

    public User(String name, String home, double height, int age, Date birthday) {
        this.name = name;
        this.home = home;
        this.height = height;
        this.age = age;
        this.birthday = birthday;
    }

    /**
     * 隨機生成一個用戶信息
     *
     * @return
     */
    public static User getOneRandomUser() {
        return new User("葫蘆" + (int) (Math.random() * 1000) + "娃", "山西省太原市" + (int) (Math.random() * 1000) + "街道", (Math.random() * 1000), (int) (Math.random() * 100), new Date(System.currentTimeMillis() - (long) (Math.random() * 100000)));
    }

    /**
     * 隨機生成num個用戶信息
     *
     * @param num 生成數量
     * @return
     */
    public static List<User> getRandomUsers(int num) {
        List<User> users = Lists.newArrayList();
        if (num < 0) num = 10;
        for (int i = 0; i < num; i++) {
            users.add(new User("葫蘆" + (int) (Math.random() * 1000) + "娃", "山西省太原市" + (int) (Math.random() * 1000) + "街道", (Math.random() * 1000), (int) (Math.random() * 100), new Date(System.currentTimeMillis() - (long) (Math.random() * 100000))));
        }

        return users;
    }

    /**
     * 封裝對象的Json信息
     *
     * @param user
     * @return
     * @throws IOException
     */
    public static XContentBuilder getXContentBuilder(User user) throws IOException {
        return XContentFactory.jsonBuilder()
                .startObject()
                .field("name", user.getName())//該字段在上面的方法中mapping定義了,因此該字段就有了自定義的屬性,好比 age等
                .field("home", user.getHome())
                .field("height", user.getHeight())
                .field("age", user.getAge())
                .field("birthday", user.getBirthday())
                .field("state", "默認屬性,mapping中沒有定義")//該字段在上面方法中的mapping中沒有定義,因此該字段的屬性使用es默認的.
                .endObject();
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getHome() {
        return home;
    }

    public void setHome(String home) {
        this.home = home;
    }

    public double getHeight() {
        return height;
    }

    public void setHeight(double height) {
        this.height = height;
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }

    public Date getBirthday() {
        return birthday;
    }

    public void setBirthday(Date birthday) {
        this.birthday = birthday;
    }

}

 

 

　　二、java與es交互demo

package com.framework_technology.elasticsearch;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequestBuilder;
import org.elasticsearch.action.admin.indices.alias.exists.AliasesExistRequestBuilder;
import org.elasticsearch.action.admin.indices.alias.exists.AliasesExistResponse;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequestBuilder;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.stats.IndexStats;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;

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

/**
 * mapping建立
 * 添加記錄到es
 */
public class Es_BuildIndex {

    /**
     * 索引的mapping
     * <p>
     * 預約義一個索引的mapping,使用mapping的好處是能夠個性的設置某個字段等的屬性
     * Es_Setting.INDEX_DEMO_01相似於數據庫
     * mapping 相似於預設某個表的字段類型
     * <p>
     * Mapping,就是對索引庫中索引的字段名及其數據類型進行定義，相似於關係數據庫中表創建時要定義字段名及其數據類型那樣，
     * 不過es的 mapping比數據庫靈活不少，它能夠動態添加字段。
     * 通常不須要要指定mapping均可以，由於es會自動根據數據格式定義它的類型，
     * 若是你須要對某 些字段添加特殊屬性（如：定義使用其它分詞器、是否分詞、是否存儲等），就必須手動添加mapping。
     * 有兩種添加mapping的方法，一種是定義在配 置文件中，一種是運行時手動提交mapping，兩種選一種就好了。
     *
     * @throws Exception Exception
     */
    protected static void buildIndexMapping() throws Exception {
        Map<String, Object> settings = new HashMap<>();
        settings.put("number_of_shards", 4);//分片數量
        settings.put("number_of_replicas", 0);//複製數量
        settings.put("refresh_interval", "10s");//刷新時間

        //在本例中主要得注意,ttl及timestamp如何用java ,這些字段的具體含義,請去到es官網查看
        CreateIndexRequestBuilder cib = Es_Utils.client.admin().indices().prepareCreate(Es_Utils.LOGSTASH_YYYY_MM_DD);
        cib.setSettings(settings);

        XContentBuilder mapping = XContentFactory.jsonBuilder()
                .startObject()
                .startObject("we3r")//
                .startObject("_ttl")//有了這個設置,就等於在這個給索引的記錄增長了失效時間,
                        //ttl的使用地方如在分佈式下,web系統用戶登陸狀態的維護.
                .field("enabled", true)//默認的false的
                .field("default", "5m")//默認的失效時間,d/h/m/s 即天/小時/分鐘/秒
                .field("store", "yes")
                .field("index", "not_analyzed")
                .endObject()
                .startObject("_timestamp")//這個字段爲時間戳字段.即你添加一條索引記錄後,自動給該記錄增長個時間字段(記錄的建立時間),搜索中能夠直接搜索該字段.
                .field("enabled", true)
                .field("store", "no")
                .field("index", "not_analyzed")
                .endObject()
                        //properties下定義的name等等就是屬於咱們須要的自定義字段了,至關於數據庫中的表字段 ,此處至關於建立數據庫表
                .startObject("properties")
                .startObject("@timestamp").field("type", "long").endObject()
                .startObject("name").field("type", "string").field("store", "yes").endObject()
                .startObject("home").field("type", "string").field("index", "not_analyzed").endObject()
                .startObject("now_home").field("type", "string").field("index", "not_analyzed").endObject()
                .startObject("height").field("type", "double").endObject()
                .startObject("age").field("type", "integer").endObject()
                .startObject("birthday").field("type", "date").field("format", "YYYY-MM-dd").endObject()
                .startObject("isRealMen").field("type", "boolean").endObject()
                .startObject("location").field("lat", "double").field("lon", "double").endObject()
                .endObject()
                .endObject()
                .endObject();
        cib.addMapping(Es_Utils.LOGSTASH_YYYY_MM_DD_MAPPING, mapping);
        cib.execute().actionGet();
    }

    /**
     * 給 []index 建立別名
     * 重載方法能夠按照過濾器或者Query 做爲一個別名
     *
     * @param aliases aliases別名
     * @param indices 多個 index
     * @return 是否完成
     */
    protected static boolean createAliases(String aliases, String... indices) {
        IndicesAliasesRequestBuilder builder = Es_Utils.client.admin().indices().prepareAliases();
        return builder.addAlias(indices, aliases).execute().isDone();
    }

    /**
     * 查詢此別名是否存在
     *
     * @param aliases aliases
     * @return 是否存在
     */
    protected static boolean aliasesExist(String... aliases) {
        AliasesExistRequestBuilder builder =
                Es_Utils.client.admin().indices().prepareAliasesExist(aliases);
        AliasesExistResponse response = builder.execute().actionGet();
        return response.isExists();
    }

    /**
     * 添加記錄到es
     * <p>
     * 增長索引記錄
     *
     * @param user 添加的記錄
     * @throws Exception Exception
     */
    protected static void buildIndex(User user) throws Exception {
        // INDEX_DEMO_01_MAPPING爲上個方法中定義的索引,prindextype爲類型.jk8231爲id,以此能夠代替memchche來進行數據的緩存
        IndexResponse response = Es_Utils.client.prepareIndex(Es_Utils.LOGSTASH_YYYY_MM_DD, Es_Utils.LOGSTASH_YYYY_MM_DD_MAPPING)
                .setSource(
                        User.getXContentBuilder(user)
                )
                .setTTL(8000)//這樣就等於單獨設定了該條記錄的失效時間,單位是毫秒,必須在mapping中打開_ttl的設置開關
                .execute()
                .actionGet();
    }

    /**
     * 批量添加記錄到索引
     *
     * @param userList 批量添加數據
     * @throws java.io.IOException IOException
     */
    protected static void buildBulkIndex(List<User> userList) throws IOException {
        BulkRequestBuilder bulkRequest = Es_Utils.client.prepareBulk();
        // either use Es_Setting.client#prepare, or use Requests# to directly build index/delete requests

        for (User user : userList) {
            //經過add批量添加
            bulkRequest.add(Es_Utils.client.prepareIndex(Es_Utils.LOGSTASH_YYYY_MM_DD, Es_Utils.LOGSTASH_YYYY_MM_DD_MAPPING)
                            .setSource(
                                    User.getXContentBuilder(user)
                            )
            );
        }

        BulkResponse bulkResponse = bulkRequest.execute().actionGet();
        //若是失敗
        if (bulkResponse.hasFailures()) {
            // process failures by iterating through each bulk response item
            System.out.println("buildFailureMessage:" + bulkResponse.buildFailureMessage());
        }
    }


}

 

數據查看、

　　經過第三方工具查看

6、執行時間

 

7、es數據增量方案

　　一、定時任務輪訓，bulk方式操做

　　　　批量添加操做：（能夠把之前的數據導進來）

　　 /**
     * 批量添加記錄到索引
     *
     * @param userList 批量添加數據
     * @throws java.io.IOException IOException
     */
    protected static void buildBulkIndex(List<User> userList) throws IOException {
        BulkRequestBuilder bulkRequest = InitES.buildClient().prepareBulk();
        // either use Es_Setting.client#prepare, or use Requests# to directly build index/delete requests

        for (User user : userList) {
            //經過add批量添加
            bulkRequest.add(InitES.buildClient().prepareIndex(LOGSTASH_YYYY_MM_DD, LOGSTASH_YYYY_MM_DD_MAPPING)
                            .setSource(
                                    User.getXContentBuilder(user)
                            )
            );
        }

        BulkResponse bulkResponse = bulkRequest.execute().actionGet();
        //若是失敗
        if (bulkResponse.hasFailures()) {
            // process failures by iterating through each bulk response item
            System.out.println("buildFailureMessage:" + bulkResponse.buildFailureMessage());
        }
    }

　　二、用隊列作數據同步，異步的方式，生產一條，放在隊列裏去消費一條。

 　　增量使用隊列，當新增一條記錄時往es中添加一條記錄

　　 protected static void buildIndex(User user) throws Exception {
        // INDEX_DEMO_01_MAPPING爲上個方法中定義的索引,prindextype爲類型.jk8231爲id,以此能夠代替memchche來進行數據的緩存
        IndexResponse response = InitES.buildClient().prepareIndex(LOGSTASH_YYYY_MM_DD, LOGSTASH_YYYY_MM_DD_MAPPING)
                .setSource(
                        User.getXContentBuilder(user)
                )
                .setTTL(8000)//這樣就等於單獨設定了該條記錄的失效時間,單位是毫秒,必須在mapping中打開_ttl的設置開關
                .execute()
                .actionGet();
    }

 

　　

 

 

　　es中的QueryBuilders的termQuery查詢，

　　   1. 若value爲漢字，則大部分狀況下，只能爲一個漢字；

        2. 若value爲英文，則是一個單詞；

　　　　queryString支持多箇中文查詢

 

 

 

TYPE:

{"mappings":{"carecustomerlog_type_all":{"properties":{"applyrate":{"type":"double"},"careAccountId":{"type":"long"},"careCustomId":{"type":"string","index":"not_analyzed"},"careaccountid":{"type":"long"},"cdate":{"type":"long"},"content":{"type":"string","index":"not_analyzed"},"customerid":{"type":"string","index":"not_analyzed"},"customername":{"type":"string","index":"not_analyzed"},"id":{"type":"long"},"orderid":{"type":"string","index":"not_analyzed"},"preapplyrate":{"type":"long"},"type":{"type":"long"},"watenum":{"type":"string","index":"not_analyzed"}}},"careCustomerLogByCustomerId":{"properties":{"customerid":{"type":"string","index":"not_analyzed"}}}}}