倒排索引

原文本表（Stored Fields）：

倒排索引表（Inverted Index）：

有了倒排索引表后，在所有文本中查找到包含目标词的文本的时间复杂度是\(O(\log N)\)。倒排索引表也不一定要存储在磁盘中，一些Term具有相同的前缀，因此可以用树结构轻量化表示所有的Term Dictionary然后存储在内存中，这就是Term Index。

Term Index：

一个Segment=Inverted Index+Term Index+Stored Fields+Doc Values。Segment的更新是通过覆盖的方式而不是修改的方式。

多个基础Segment就可以组成一个基础搜索库Lucene，ElasticSearch就是基于Lucene。

在ES中，一个Shard就是一个Lucene库，多个Shard可以分布式运行在不同Node上，因此也诞生了Shard副本概念（和Kafka的分区相似）。大规模分布式ES中，有三类Node分工：1. Master Node（集群管理）；2. Data Node（存储数据）；3. Coordinate Node（处理请求）。

ES的搜索流程：

1. Query Phase：客户端请求 -> Coordinate Node -> 多个Data Node -> 并发搜索每个Segment -> 返回文档ID和排序信息（？）

2. Fetch Phase：Coordinate Node 根据文档ID -> 多个Data Node -> 检索Stored Fields -> 返回目标文档

Elasticsearch安装

1. 下载安装包：

wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-8.14.3-linux-x86_64.tar.gz

2. 在新建用户zt的.profile文件中追加下述内容以配置环境变量。

vim /home/zt/.profile

export ES_JAVA_HOME=/home/zt/Elasticsearch814/elasticsearch-8.14.3/jdk/
export ES_HOME=/home/zt/Elasticsearch814/elasticsearch-8.14.3

执行source .profile

3. 修改配置

vim $ES_HOME/config/elasticsearch.yml

修改：

# 使本地电脑能够访问到远程服务器
network.host: 0.0.0.0
# 指定为单节点，绕过引导检查，设置为开发者模式
discovery.type: single-node
# 开启security安全认证
xpack.security.enabled: true

vim $ES_HOME/config/jvm.options

修改下述内容：我设成了1G

-Xms1g
-Xmx1g

为Elasticsearch配置用户密码

参考连接

在保持ES启动的情况下使用下述所有命令。

# 查询ES中现有的用户
$ES_HOME/bin/elasticsearch-users list
# 新建ES用户zt并赋予超级用户权限，然后设置密码
$ES_HOME/bin/elasticsearch-users useradd zt -r superuser

但是通过上述命令自建的用户zt会导致连接Kibana失败。所以还是通过下述命令为系统用户apm_system、kibana_system、kibana、logstash_system、beats_system、remote_monitoring_user、elastic生成密码。但是这些密码似乎只会在console中被打印出一次，所以应该妥善保存。

$ES_HOME/bin/elasticsearch-setup-passwords auto

Elasticsearch启动和停止

后台启动：

$ES_HOME/bin/elasticsearch -d

停止：

ps -ef | grep elasticsearch | grep -v grep | awk '{print $2}' | xargs kill

之后浏览器访问：http://公网IP:9200即可看见结果。Chrome可以安装multi elasticsearch heads插件美化可视化结果。

分词器安装

这里下载ICU分词器。

$ES_HOME/bin/elasticsearch-plugin install analysis-icu

再下载一个分词器：

wget https://release.infinilabs.com/analysis-ik/stable/elasticsearch-analysis-ik-8.14.3.zip
unzip elasticsearch-analysis-ik-8.14.3.zip -d $ES_HOME/plugins/elasticsearch-analysis-ik-8.14.3

安装完毕后重启ES。

在浏览器打开的网页的的符合查询栏目，对http://公网IP:9200/_analyze发送POST请求，请求体如下。点击“提交请求”可以看到分词结果。

{
  "analyzer": "icu_analyzer",
  "text": "你好，我的Elasticsearch"
}

Kibana安装和配置

wget https://artifacts.elastic.co/downloads/kibana/kibana-8.14.3-linux-x86_64.tar.gz

vim /home/zt/Kibana814/kibana-8.14.3/config/kibana.yml

server.host: "0.0.0.0"
i18n.locale: "zh-CN"
elasticsearch.hosts: ["http://192.168.0.45:9200"]
elasticsearch.username: "kibana_system"
elasticsearch.password: "kibana_system的密码"

nohup /home/zt/Kibana814/kibana-8.14.3/bin/kibana &

ps -ef | grep kibana | grep -v grep | awk '{print $2}' | xargs kill

Kibana用法

打开“开发工具”

# 创建索引myindex
PUT /myindex
# 获取索引myindex
GET /myindex
# 在索引myindex中创建映射
POST /myindex/_mapping
{
    "properties": {
        "content": {
            "type": "text",
            "analyzer": "ik_max_word",
            "search_analyzer": "ik_smart"
        }
    }
}
# 往索引myindex添加数据
POST /myindex/_create/0
{
    "content": "你好，Elasticsearch。你好，西伯利亚蝴蝶"
}
POST /myindex/_create/1
{
    "content": "你好，Elasticsearch。你好，仙女座星系"
}
# 在myindex索引中进行搜索
GET /myindex/_search
{
    "query": {"match": {"content": "你好"}}
}
# 删除索引myindex
DELETE /myindex
# 修改副本数量
PUT /myindex/_settings
{
    "index": {
        "number_of_replicas: 2"
    }
}
# 新增字段
PUT /myindex/_mapping
{
  "properties": {
    "grade": {
      "type": "integer"
    }
  }
}

环境准备

在Kibana中运行GET /来查看ES版本号（位于”version”->“number”）中。例如我的当前ES版本为8.14.3。

pip install elasticsearch==8.17.0 

from elasticsearch import Elasticsearch
es = Elasticsearch("http://公网IP:9200")
print(es.info())  # 能够打印出信息说明本地连接Elasticsearch成功

Python ES 查询语句

query = {
  "query": {
    "bool": {
      "must": [
        {"match": {"mapping1": "value0"}},
        {"match": {"mapping2": "value1"}},
        {"match": {"mapping3": "value2"}}
        ]
    }
  }
}
query_result = es.count(index="my_index_name", body=search_query)
print(query_result)  # 打印出信息{'count': 261285, '_shards': {'total': 3, 'successful': 3, 'skipped': 0, 'failed': 0}}
query_result = es.search(index="my_index_name", body=search_query)
print(query_result)  # 打印出查询到的所有信息

文本向量化

借助阿里云百炼平台。

from openai import OpenAI

llm_client = OpenAI(api_key=<API_KEY>, base_url='https://dashscope.aliyuncs.com/compatible-mode/v1')

text = "Hello, ElasticSearch"
tmp = llm_client.embeddings.create(input=text,  model='text-embedding-v3')

# 打印出文本向量化后的结果以及向量维度。这里应该是(1, 1024)维度
print(tmp.data[0].embedding, len(tmp.data[0].embedding))

文本向量入库

from elasticsearch import Elasticsearch

es = Elasticsearch("http://公网IP:9200", request_timeout=30)
print(es.info())  # 能打印出结果则连接成功

1. 创建索引用于存储文本向量。

mappings = {
    "mappings": {
        "properties": {
            "semantic": {
                "type": "dense_vector",
                "dims": 1024
            },
            "content": {
                "type": "text"
            }
        }
    }
}
es.indices.create(index='my_rag_vector_index', body=mappings)

senmantic的类型dense_vector在ElasticSearch 7 版本中不受支持。

2. 向量文本存入新构建的索引

# documnets格式：[{'text': ..., 'text_vector': ...}, ...]
def generate_docs(documents):
    index_name = 'my_rag_vector_index'
    result = []
    for row in documents:
        text = row['text']
        text_vector = row['text_vector']
        doc = {'content': text, 'semantic': text_vector}
        result.append({'index': {'_index': index_name}})
        result.append(doc)
    return result
    
es.bulk(body=generate_docs(documents)) 

检索

user_text = 'ElasticSearch'
user_text_vector = llm_client.embeddings.create(input=text, model='text-embedding-v3').data[0].embedding
query = {
    'knn': {
        'field': 'semantic',
        'query_vector': user_text_vector,
        'num_candidates': 1
    }
}
result = es.search(index='rag_vector_index_zt', body={'query': query})
print(result)