Elasticsearch——倒排索引与分词

2020-11-09 本文已影响0人小波同学

正排索引

文档ID到文档内容、单词的关联关系。比如书的目录页对应正排索引（指明章节名称，指明页数）用于查看章节。

倒排索引

单词到文档ID的关联关系。比如索引页对应倒排索引(指明关键词、指明页数)用于关键词查找
倒排索引是搜索引擎的核心，主要包含两个部分：
单词词典(Term Dictionary)

记录所有文档的单词，一般都比较大。
记录单词到倒排列表的关联信息。

倒排列表(Posting List)
记录了单词对应的文档集合，由倒排索引项组成。倒排索引项包含如下信息：

文档ID，用于获取原始信息。
单词频率，记录该单词在该文档中的出现次数，用于后续相关性算分。
位置，记录单词在文档中的粉刺位置，用于做词语搜索。
偏移，记录单词在文档的开始和结束位置，用于做高亮显示。

分词

分词是指将文本转换成一系列单词的过程，也可以叫做文本分析，在es里面成为Analysis

分词器是Elasticsearch中专门处理分词的组件，英文为Analyzer，其组成如下：
Character Filters
针对原始文本进行处理，比如去除html特殊标记符。

Tokenizer
将原始文本按照一定规则切分为单词。

Token Filters
针对Tokenizer处理的单词进行在加工，比如转小写，删除或新增等处理。

分词器——调用顺序

Analyze_api

Elasticsearch提供了一个测试分词的api接口，方便验证分词效果，endpoint是_analyze

可以直接指定Analyzer进行测试。
可以直接指定索引中的字段进行测试。
可以自定义分词器进行测试。
直接指定analyze进行测试，接口如下：
直接指定索引中的字段进行测试，接口如下：
自定义分词器进行测试，接口如下：

Elasticsearch自带分词器

中文分词

难点：

中文分词指的是将一个汉字序列切分成一个一个单独的词，在英文中单词之间是以空格作为自然分隔符，但汉语中则没有形式上的分隔符。
上下文不同分词效果迥异，比如交叉歧义问题，比如下面两种分词都合理。

乒乓球拍/卖/完了
乒乓球/拍/买完了

常用分词系统

实现中英文单词的切分，支持ik_smart、ik_maxword等模式
可自定义词库，支持热更新分词词典
https://github.com/medcl/elasticsearch-analysis-ik
下载地址：https://github.com/medcl/elasticsearch-analysis-ik/releases

ieba

python中最流行的分词系统，支持分词和词性标注
支持繁体分词，自定义词典，并行分词等
https://github.com/sing1ee/elasticsearch-jieba-plugin

基于自然语言处理的分词系统

HanLp

由一系列模型与算法组成的Java工具包，目标是普及自然语言处理在生产环境中的应用
https://github.com/hankcs/HanLp

thulac

THU Lexical Analyzer for Chinese，由清华大学自然语言处理与社会人文计算实验室研制推出的一套中文词法分析工具包，具有中文分词和词性标注功能
https://github.com/microbun/elasticsearch-thulac-plugin

自定义分词

当自带的分词无法满足需求时，可自定义分词
通过自定义Character Filters、Tokenizer、Token Filters实现

Character Filters

在Tokenizer之前对原始文本进行处理，比如增加、删除或替换字符等。
自带的如下：
- HTML Strip 去除html标签和转换html实体。
- Mapping进行字符替换操作。
- Pattern Replace进行正则匹配替换。
会影响后续Tokenizer解析的postion和offset信息。

Tokenizer

将原始文本按照一定规则切分为单词(term or token)
自带的如下：
- standard 按照单词进行分割。
- letter 按照非字符类进行分割。
- whitespace 按照空格进行分割。
- UAX URL Email 按照standard 分割，但不会分割邮箱和url。
- NGram和Edge NGram连词分割。
- Path Hierarchy 按照文件路径进行分割。

Token Filters

对于Tokenizer输出的单词(term)进行增加、删除、修改等操作。
自带的如下：
- lowercase 将所有的term转换为小写。
- stop删除stop words。
- NGram和Edge NGram连词分割。
- Synonym添加近义词的term。

自定义分词的api

自定义分词需要在索引的配置中设定，如下所示：

分词会在如下两个时机使用：

创建或更新文档时(Index Time)，会对相应的文档进行分词处理。
查询时(Search Time)，会对查询语句进行分词。

索引时分词是通过配置Index Mapping中每个字段的analyzer属性实现的，如下：

不指定分词时，使用默认分词standard

查询时分词的指定方式有如下几种：

查询的时候通过analyzer指定分词器。
通过index mapping设置search_analyzer实现

ik分词器安装与使用

ik分词器下载与安装

1、下载与ES对应的版本即可，下载地址：https://github.com/medcl/elasticsearch-analysis-ik/releases

*2、解压，将文件复制到es安装目录/plugins/ik目录下即可

3、重启elasticsearch

ik分词器基础知识

ik_max_word：会将文本做最细粒度的拆分，比如会将"中华人民共和国人民大会堂"拆分为"中华人民共和国、中华人民、中华、华人、人民、人民共和国、人民大会堂、人民大会、大会堂"，会穷尽各种可能的组合。

ik_smart：会做最粗粒度的拆分，比如会将"中华人民共和国人民大会堂"拆分为"中华人民共和国、人民大会堂"。

ik分词器的使用

存储时使用ik_max_word，搜索时使用ik_smart

因为后续的keyword和text设计分词问题，这里给出分词最佳实践。即存储时时使用ik_max_word，搜索时分词器用ik_smart，这样索引时最大化的将内容分词，搜索时更精确的搜索到想要的结果。

PUT /index

{
    "mappings": {
        "peoperties":{
            "text":{
                "type": "text",
                "analyzer": "ik_max_word",
                "search_analyzer": "ik_smart"
            }
        }
    }
}

ik分词器配置文件

ik配置文件地址/es/plugins/ik/config目录
IKAnalyzer.cfg.xml：用来配置自定义词库。
main.dic：ik原生内置的中文词库，总共有27万多条，只要是这些单词，都会被分到一起。
preposition.dic：介词。
quantifier.dic：放了一些单位相关的词，量词。
suffix.dic：放了一些后缀。
surname.dic：中国的姓氏。
stopword.dic：英文停用词。

ik原生最重要的两个配置文件：

main.dic：包含了原生的中文词语，会按照这里面的词去进行分词。
stopword.dic：包含了英文的停用词。

一般向停用词会在分词的时候，直接被干掉，不会建立在倒排索引中。

自定义词库

1、自己建立词库：每年都会涌现一些特殊的流行词，网红、蓝瘦香菇、喊麦、鬼畜等，一般不会在ik的原生词典里。
自己补充自己的最新的词语，到ik词库里面。
在IKAnalyzer.cfg.xml文件中ext_dict标签中，创建mydict.dic。

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE properties SYSTEM "http://java.sun.com/dtd/properties.dtd">
<properties>
    <comment>IK Analyzer 扩展配置</comment>
    <!--用户可以在这里配置自己的扩展字典 -->
    <entry key="ext_dict">mydict.dic</entry>
     <!--用户可以在这里配置自己的扩展停止词字典-->
    <entry key="ext_stopwords">mystopwords.dic</entry>
    <!--用户可以在这里配置远程扩展字典 -->
    <!-- <entry key="remote_ext_dict">words_location</entry> -->
    <!--用户可以在这里配置远程扩展停止词字典-->
    <!-- <entry key="remote_ext_stopwords">words_location</entry> -->
</properties>

2、自己建立停用词库，比如了、的、啥、么等，可能并不想建立索引让人家搜索。
custom/ext_stopword.dic，已经有了常用的中文停用词，可以自己补充自己的中文停用词，然后重启es。

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE properties SYSTEM "http://java.sun.com/dtd/properties.dtd">
<properties>
    <comment>IK Analyzer 扩展配置</comment>
    <!--用户可以在这里配置自己的扩展字典 -->
    <entry key="ext_dict">mydict.dic</entry>
     <!--用户可以在这里配置自己的扩展停止词字典-->
    <entry key="ext_stopwords">mystopwords.dic</entry>
    <!--用户可以在这里配置远程扩展字典 -->
    <!-- <entry key="remote_ext_dict">words_location</entry> -->
    <!--用户可以在这里配置远程扩展停止词字典-->
    <!-- <entry key="remote_ext_stopwords">words_location</entry> -->
</properties>

使用mysql热更新词库

热更新

每次都是在es的扩展词典中，手动添加新词，很坑。

每次添加完，都要重启es才能生效，非常麻烦。
es是分布式的可能有数百个节点，你不能每一次都一个一个节点上面去修改。

es不停机，我们直接在外部某个地方添加新的词语，es中立即热加载到这些新词语。

热更新方案

1、基于ik分词器原生支持的热更新方案，部署一个web服务器，提供一个http接口，通过modified和tag两个http响应头，来提供词语的热更新。
2、修改ik分词器源码，然后手动支持从mysql中每隔一段时间，自动加载新的词库。

用第二种方案，第一种方案ik官方和社区都不建议采用，觉得不太稳定。

1、基于ik分词器原生支持的热更新方案

1)、ik官方文档说明
目前该插件支持热更新 IK 分词，通过上文在 IK 配置文件中提到的如下配置

<!--用户可以在这里配置远程扩展字典 -->
<entry key="remote_ext_dict">location</entry>
<!--用户可以在这里配置远程扩展停止词字典-->
<entry key="remote_ext_stopwords">location</entry>

其中 location 是指一个 url，比如 http://yoursite.com/getCustomDict，该请求只需满足以下两点即可完成分词热更新。

A、该 http 请求需要返回两个头部(header)，一个是 Last-Modified，一个是 ETag，这两者都是字符串类型，只要有一个发生变化，该插件就会去抓取新的分词进而更新词库。
B、该 http 请求返回的内容格式是一行一个分词，换行符用 \n 即可。

满足上面两点要求就可以实现热更新分词了，不需要重启 ES 实例。

可以将需自动更新的热词放在一个 UTF-8 编码的xxx.txt文件里，放在 nginx 或其他简易 http server下，当xxx.txt文件修改时，http server 会在客户端请求该文件时自动返回相应的 Last-Modified 和 ETag。可以另外做一个工具来从业务系统提取相关词汇，并更新这个xxx.txt文件。

个人体会：nginx方式比较简单容易实现，建议使用；

2)、在服务中实现http请求，并连接数据库实现热词管理实例：
A、编写http请求服务接口demo

@RestController
@RequestMapping("/keyWord")
@Slf4j
public class KeyWordDict {

    private String lastModified = new Date().toString();
    private String etag = String.valueOf(System.currentTimeMillis());

    @RequestMapping(value = "/hot", method = {RequestMethod.GET,RequestMethod.HEAD}, produces="text/html;charset=UTF-8")
    public String getHotWordByOracle(HttpServletResponse response,Integer type){
        response.setHeader("Last-Modified",lastModified);
        response.setHeader("ETag",etag);

        Connection conn = null;
        Statement stmt = null;
        ResultSet rs = null;
        String sql = "";
        final ArrayList<String> list = new ArrayList<String>();
        StringBuilder words = new StringBuilder();
        try {
            Class.forName("oracle.jdbc.driver.OracleDriver");
            conn = DriverManager.getConnection(
                    "jdbc:oracle:thin:@192.168.114.13:1521:xe",
                    "test",
                    "test"
            );
            if(ObjectUtils.isEmpty(type)){
                type = 99;
            }
            switch (type){
                case 0:
                    sql = "select word from IK_HOT_WORD where type=0 and status=0";
                    break;
                case 1:
                    sql = "select word from IK_HOT_WORD where type=1 and status=0";
                    break;
                default:
                    sql = "select word from IK_HOT_WORD where type=99";
                    break;
            }
            stmt = conn.createStatement();
            rs = stmt.executeQuery(sql);

            while(rs.next()) {
                String theWord = rs.getString("word");
                System.out.println("hot word from mysql: " + theWord);
                words.append(theWord);
                words.append("\n");
            }
            return words.toString();

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if(rs != null) {
                try {
                    rs.close();
                } catch (SQLException e) {
                    log.error("资源关闭异常：",e);
                }
            }
            if(stmt != null) {
                try {
                    stmt.close();
                } catch (SQLException e) {
                    log.error("资源关闭异常：",e);
                }
            }
            if(conn != null) {
                try {
                    conn.close();
                } catch (SQLException e) {
                    log.error("资源关闭异常：",e);
                }
            }
        }
        return null;
    }

    @RequestMapping(value = "/update", method = RequestMethod.GET)
    public void updateModified(){
        lastModified = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss").format(new Date());
        etag = String.valueOf(System.currentTimeMillis());
    }

}

注：
updateModified方法为单独更新lastModified与etag，用于判断ik是否需要重新加载远程词库，具体关联数据库操作代码时自行扩展

B、ik配置文件修改
- 文件目录：/data/elasticsearch-7.3.0/plugins/ik/config/IKAnalyzer.cfg.xml
- 远程调用方法填写在“用户可以在这里配置远程扩展字典”下：

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE properties SYSTEM "http://java.sun.com/dtd/properties.dtd">
<properties>
        <comment>IK Analyzer 扩展配置</comment>
        <!--用户可以在这里配置自己的扩展字典 -->
        <entry key="ext_dict"></entry>
         <!--用户可以在这里配置自己的扩展停止词字典-->
        <entry key="ext_stopwords"></entry>
        <!--用户可以在这里配置远程扩展字典 -->
        <entry key="remote_ext_dict">http://192.168.xx.xx:8080/keyWord/hot?type=0</entry>
        <!--用户可以在这里配置远程扩展停止词字典-->
        <!-- <entry key="remote_ext_stopwords">words_location</entry> -->
</properties>

重写ik源码连接mysql/oracle更新词库

1、下载ik源码(下载对应版本)
https://github.com/medcl/elasticsearch-analysis-ik/releases
ik分词器是一个标准的java maven工程，直接导入idea就可看到源码。
2、修改ik插件源码（以mysql为例）
1）、添加jdbc配置文件
在项目根目录下的config目录中添加config\jdbc-reload.properties配置文件：

jdbc.driverClass=com.mysql.jdbc.Driver
jdbc.url=jdbc:mysql://localhost:3306/es?serverTimezone=UTC
jdbc.user=root
jdbc.password=yibo
jdbc.reload.sql=select word from hot_words
jdbc.reload.stopword.sql=select stopword as word from hot_stopwords
jdbc.reload.interval=5000

2）、在Dictionary类的同级目录下新建HotDictReloadThread类

/**
 * @Description: 加载字典线程
 */
public class HotDictReloadThread implements Runnable {

    private static final Logger log = ESPluginLoggerFactory.getLogger(HotDictReloadThread.class.getName());

    @Override
    public void run() {
        log.info("[--------]reload hot dict from mysql");
        Dictionary.getSingleton().reLoadMainDict();
    }
}

3）、在Dictionary类initial方法中新增代码

public static synchronized void initial(Configuration cfg) {
    if (singleton == null) {
        synchronized (Dictionary.class) {
            if (singleton == null) {

                singleton = new Dictionary(cfg);
                singleton.loadMainDict();
                singleton.loadSurnameDict();
                singleton.loadQuantifierDict();
                singleton.loadSuffixDict();
                singleton.loadPrepDict();
                singleton.loadStopWordDict();

                //!!!!!!!!mysql监控线程  新增代码
                new Thread(new HotDictReloadThread()).start();

                if(cfg.isEnableRemoteDict()){
                    // 建立监控线程
                    for (String location : singleton.getRemoteExtDictionarys()) {
                        // 10 秒是初始延迟可以修改的 60是间隔时间 单位秒
                        pool.scheduleAtFixedRate(new Monitor(location), 10, 60, TimeUnit.SECONDS);
                    }
                    for (String location : singleton.getRemoteExtStopWordDictionarys()) {
                        pool.scheduleAtFixedRate(new Monitor(location), 10, 60, TimeUnit.SECONDS);
                    }
                }

            }
        }
    }
}

4）、在Dictionary类loadMainDict方法中新增代码

/**
 * 加载主词典及扩展词典
 */
private void loadMainDict() {
    // 建立一个主词典实例
    _MainDict = new DictSegment((char) 0);

    // 读取主词典文件
    Path file = PathUtils.get(getDictRoot(), Dictionary.PATH_DIC_MAIN);
    loadDictFile(_MainDict, file, false, "Main Dict");
    // 加载扩展词典
    this.loadExtDict();
    // 加载远程自定义词库
    this.loadRemoteExtDict();
    //从mysql中加载热更新词典 新增代码
    this.loadMySQLExtDict();
}

5）、在Dictionary类loadStopWordDict方法中新增代码

/**
 * 加载用户扩展的停止词词典
 */
private void loadStopWordDict() {
    // 建立主词典实例
    _StopWords = new DictSegment((char) 0);

    // 读取主词典文件
    Path file = PathUtils.get(getDictRoot(), Dictionary.PATH_DIC_STOP);
    loadDictFile(_StopWords, file, false, "Main Stopwords");

    // 加载扩展停止词典
    List<String> extStopWordDictFiles = getExtStopWordDictionarys();
    if (extStopWordDictFiles != null) {
        for (String extStopWordDictName : extStopWordDictFiles) {
            logger.info("[Dict Loading] " + extStopWordDictName);

            // 读取扩展词典文件
            file = PathUtils.get(extStopWordDictName);
            loadDictFile(_StopWords, file, false, "Extra Stopwords");
        }
    }

    // 加载远程停用词典
    List<String> remoteExtStopWordDictFiles = getRemoteExtStopWordDictionarys();
    for (String location : remoteExtStopWordDictFiles) {
        logger.info("[Dict Loading] " + location);
        List<String> lists = getRemoteWords(location);
        // 如果找不到扩展的字典，则忽略
        if (lists == null) {
            logger.error("[Dict Loading] " + location + "加载失败");
            continue;
        }
        for (String theWord : lists) {
            if (theWord != null && !"".equals(theWord.trim())) {
                // 加载远程词典数据到主内存中
                logger.info(theWord);
                _StopWords.fillSegment(theWord.trim().toLowerCase().toCharArray());
            }
        }
    }

    //!!!!!!!!从mysql中加载停用词 新增代码
    this.loadMySQLStopWordDict();
}

6）、在Dictionary类新增静态代码块，加载db驱动

private static Properties prop = new Properties();

static {
    try {
        Class.forName("com.mysql.cj.jdbc.Driver");
    } catch (ClassNotFoundException e) {
        logger.error("error",e);
    }
}

7）、在Dictionary类新增loadMySQLExtDict方法，从mysql中加载热更新词典

/**
 * 从mysql中加载热更新词典
 */
private void loadMySQLExtDict(){
    Connection conn = null;
    Statement state = null;
    ResultSet rs = null;
    try{
        Path file = PathUtils.get(getDictRoot(),"jdbc-reload.properties");
        prop.load(new FileInputStream(file.toFile()));
        for (Object key : prop.keySet()) {
            logger.info("[--------]" + key +"=" + prop.getProperty(String.valueOf(key)));
        }
        logger.info("[--------]query hot dict from mysql," + prop.getProperty("jdbc.reload.sql") + "......");

        // 创建数据连接
        conn = DriverManager.getConnection(
                prop.getProperty("jdbc.url"),
                prop.getProperty("jdbc.user"),
                prop.getProperty("jdbc.password")
        );

        state = conn.createStatement();
        rs = state.executeQuery(prop.getProperty("jdbc.reload.sql"));

        while(rs.next()){
            String theWord = rs.getString("word");
            logger.info("[--------]hot word from mysql: " + theWord);
            _MainDict.fillSegment(theWord.trim().toCharArray());
        }

        Thread.sleep(Integer.valueOf(String.valueOf(prop.get("jdbc.reload.interval"))));
    }catch (Exception e){
        logger.error("error",e);
    }finally {
        if(rs != null){
            try {
                rs.close();
            } catch (SQLException e) {
                logger.error("error",e);
            }
        }
        if(state != null){
            try {
                state.close();
            } catch (SQLException e) {
                logger.error("error",e);
            }
        }
        if(conn != null){
            try {
                conn.close();
            } catch (SQLException e) {
                logger.error("error",e);
            }
        }
    }
}

8）、在Dictionary类新增loadMySQLStopWordDict方法，从mysql中加载停用词

/**
 * 从mysql中加载停用词
 */
private void loadMySQLStopWordDict(){
    Connection conn = null;
    Statement state = null;
    ResultSet rs = null;
    try{
        Path file = PathUtils.get(getDictRoot(),"jdbc-reload.properties");
        prop.load(new FileInputStream(file.toFile()));
        for (Object key : prop.keySet()) {
            logger.info("[--------]" + key +"=" + prop.getProperty(String.valueOf(key)));
        }
        logger.info("[--------]query hot stopword from mysql," + prop.getProperty("jdbc.reload.stopword.sql") + "......");

        // 创建数据连接
        conn = DriverManager.getConnection(
                prop.getProperty("jdbc.url"),
                prop.getProperty("jdbc.user"),
                prop.getProperty("jdbc.password")
        );

        state = conn.createStatement();
        rs = state.executeQuery(prop.getProperty("jdbc.reload.stopword.sql"));

        while(rs.next()){
            String theWord = rs.getString("word");
            logger.info("[--------]hot stopword from mysql: " + theWord);
            _MainDict.fillSegment(theWord.trim().toCharArray());
        }

        Thread.sleep(Integer.valueOf(String.valueOf(prop.get("jdbc.reload.interval"))));
    }catch (Exception e){
        logger.error("error",e);
    }finally {
        if(rs != null){
            try {
                rs.close();
            } catch (SQLException e) {
                logger.error("error",e);
            }
        }
        if(state != null){
            try {
                state.close();
            } catch (SQLException e) {
                logger.error("error",e);
            }
        }
        if(conn != null){
            try {
                conn.close();
            } catch (SQLException e) {
                logger.error("error",e);
            }
        }
    }
}

3、mvn package打包代码
4、解压ik压缩包
将mysql驱动jar，放入ik目录下。
5、修改jdbc相关配置
6、重启es
观察日志，日志中会显示打印那些东西，比如加载了什么配置，加载了什么词语，加载了什么停用词。
7、在MySQL中添加词库与停用词
8、分词验证，验证热更新

总结：

一般不需要特别指定查询时分词器，直接使用索引时分词器即可，否则会出现无法匹配的情况。
明确字段是否需要分词，不需要分词的字段就将type设置为keyword，可以节省空间和提高写性能。
善用_analyze API，查看文档的具体分词结果。

参考：
https://github.com/medcl/elasticsearch-analysis-ik

https://blog.csdn.net/qq_40592041/article/details/107856588