首页 > 其他分享 >Rust整合Elasticsearch

Rust整合Elasticsearch

时间：2024-10-28 17:50:31浏览次数：8  
标签：unwrap client let 整合 test Elasticsearch config response Rust

Elasticsearch是什么

Lucene：Java实现的搜索引擎类库

  • 易扩展
  • 高性能
  • 仅限Java开发
  • 不支持水平扩展

Elasticsearch：基于Lucene开发的分布式搜索和分析引擎

  • 支持分布式、水平扩展
  • 提高RestfulAPI，可被任何语言调用

Elastic Stack是什么

ELK（Elastic Stack）：Elasticsearch结合Kibana、Logstash、Beats实现日志数据分析、实时监控

  • Elasticsearch：负责存储、搜索、分析数据
  • Kibana：数据可视化
  • Logstash、Beats：数据抓取（一般用Debezium、Flink、RisingWave…）

Elasticsearch能做什么

  • 实时数据分析：支持对实时数据进行索引和分析，可快速处理大量的日志、指标和事件数据
  • 实时监控：对系统指标、业务数据和用户行为进行实时监控
  • 电商搜索：为电商平台提供商品搜索功能，帮助用户快速找到所需的商品
  • 知识库搜索：为企业内部的文档、知识库和业务数据提供搜索功能，提高员工的工作效率

Elasticsearch 索引

传统数据库使用正向索引，依据id构建B+树，根据索引id查快，对于非索引文档如商品描述查需要全表扫描

倒排索引：将文档分为词条和id进行存储，先查文档获取id，再根据id查数据库

  • 文档（Document）：每条数据就是一个Json文档
  • 词条（Term）：文档按语义分成的词语

索引（Index）：相同类型文档的集合
映射（Mapping）：索引中的文档约束信息
字段（Fielf）：Json文档中的字段
DSL：Json风格的请求语句，用来实现CRUD

Docker安装Elasticsearch、Kibana、IK

1、先创建自定义网络

使用默认bridge只能通过ip通信，这里加入了自定义网络，自定义网络可以自动解析容器名

  • docker network ls查看已有网络
  • 创建自定义网络docker network create pub-network
  • 手动连接网络docker network connect pub-network container_name_or_id
  • 删除网络docker network rm network_name_or_idid

2、创建文件夹

mkdir -p /opt/es/data

mkdir -p /opt/es/plugins

mkdir -p /opt/es/logs

3、授权

chmod -R 777 /opt/es/data

chmod -R 777 /opt/es/logs

安装IK分词器

由于ES对中文分词无法理解语义，需要IK插件
https://release.infinilabs.com/analysis-ik/stable/

Elasticsearch、Kibana、IK所有版本保持一致，解压后使用shell工具将整个文件夹上传到/opt/es/plugins

离线部署Elasticsearch、Kibana

在能访问的地方拉取镜像

docker pull elasticsearch:8.15.2

docker pull kibana:8.15.2

这里使用wsl，wsl进入wsl，然后进入win的D盘

cd /mnt/d

打包镜像，这个文件可以在win D盘找到

docker save elasticsearch:8.15.2 > elasticsearch.tar

docker save kibana:8.15.2 > kibana.tar

使用shell工具如Windterm上传文件

加载镜像

docker load -i elasticsearch.tar

docker load -i kibana.tar

查看镜像

docker images

然后命令部署或者docker-compose部署即可

命令部署Elasticsearch、Kibana

部署Elasticsearch

docker run -d \
--name es \
--network pub-network \
--restart always \
-p 9200:9200 \
-p 9300:9300 \
-e "xpack.security.enabled=false" \
-e "discovery.type=single-node" \
-e "http.cors.enabled=true" \
-e "http.cors.allow-origin:*" \
-e "ES_JAVA_OPTS=-Xms512m -Xmx512m" \
-v /opt/es/data:/usr/share/elasticsearch/data \
-v /opt/es/plugins:/usr/share/elasticsearch/plugins \
-v /opt/es/logs:/usr/share/elasticsearch/logs \
--privileged=true \
elasticsearch:8.15.2

xpack.security.enabled=false禁用密码登录

如果要使用token: -e "xpack.security.enrollment.enabled=true" \

docker部署一般用于开发，不要为难自己，使用token会有很多问题，生产环境再开，使用SSl需要证书

部署Kibana

docker run -d \
--name kibana \
--network pub-network \
--restart always \
-p 5601:5601 \
-e CSP_STRICT=false \
-e I18N_LOCALE=zh-CN \
kibana:8.15.2

报错kibana 服务器尚未准备就绪，是因为配置了ELASTICSEARCH_HOSTS

docker-compose部署Elasticsearch、Kibana

  es:
    image: elasticsearch:8.15.2
    container_name: es
    network_mode: pub-network
    restart: always
    ports:
      # 9200：对外暴露的端口
      - 9200:9200
      # 9300：节点间通信端口
      - 9300:9300
    environment:
      # 禁用密码登录
      xpack.security.enabled: 'false'
      # 单节点运行
      discovery.type: single-node
      # 允许跨域
      http.cors.enabled: 'true'
      # 允许所有访问
      http.cors.allow-origin: '*'
      # 堆内存大小
      ES_JAVA_OPTS: '-Xms512m -Xmx512m'
    volumes:
      # 数据挂载
      - /opt/es/data:/usr/share/elasticsearch/data
      # 插件挂载
      - /opt/es/plugins:/usr/share/elasticsearch/plugins
      # 日志挂载
      - /opt/es/logs:/usr/share/elasticsearch/logs
    # 允许root用户运行
    privileged: true
  kibana:
    image: kibana:8.15.2
    container_name: kibana
    network_mode: pub-network
    restart: always
    ports:
      - 5601:5601
    environment:
      # 禁用安全检查
      CSP_STRICT: 'false'
      # 设置中文
      I18N_LOCALE: zh-CN
networks:
  pub-network:
    name: pub-network

部署

docker-compose up -d

删除Elasticsearch、Kibana

docker rm -f es

docker rm -f kibana

开启安全配置（可选，如果要用密码和token）

es8开始需要密码访问，kibana通过token访问

# 生成密码
docker exec -it es /usr/share/elasticsearch/bin/elasticsearch-reset-password -u elastic
# 生成kibana访问token
docker exec -it es /usr/share/elasticsearch/bin/elasticsearch-create-enrollment-token -s kibana

访问Elasticsearch、Kibana

Elasticsearch：127.0.0.1:9200，看到以下界面就部署成功了

在这里插入图片描述

Kibana：127.0.0.1:5601看到以下界面就部署成功了

访问：http://127.0.0.1:9200/.kibana跨域查看有没有发现可视化工具kibana

在这里插入图片描述

我们选择手动配置，使用http://es:9200，我们没有配置ssl只能用http，容器名为es

在这里插入图片描述

在终端运行命令查看日志中的验证码

docker logs kibana

在这里插入图片描述

使用

在这里插入图片描述

GET /_analyze
{
  "analyzer": "ik_max_word",
  "text": "好好学习天天向上"
}

如果一个字为一个词条，就说明分词插件IK没装好，重新安装后重启容器docker restart es

在这里插入图片描述

分词原理

依据字典进行分词

对于一些新词语，如铝合金键盘被称为“铝坨坨”，词典中没有这个词语，会将其逐字分词
在这里插入图片描述

分词流程

  • 1、character filters：字符过滤器，进行原始处理，如转换编码、去停用词、转小写
  • 2、tokenizer：分词器，将文本流进行分词为词条
  • 3、tokenizer filter：将词条进行进一步处理，如同义词处理、拼音处理

扩展词库

在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">ext.dic</entry>
	 <!--用户可以在这里配置自己的扩展停止词字典-->
	<entry key="ext_stopwords">stopword.dic</entry>
	<!--用户可以在这里配置远程扩展字典 -->
	<!-- <entry key="remote_ext_dict">words_location</entry> -->
	<!--用户可以在这里配置远程扩展停止词字典-->
	<!-- <entry key="remote_ext_stopwords">words_location</entry> -->
</properties>

停用词库

例如敏感词

<?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_stopwords">stopword.dic</entry>
</properties>

使用

生产使用可以用AI、ELP进行分词

修改配置，添加扩展词库和停用词库

vim /opt/es/plugins/elasticsearch-analysis-ik-8.15.2/config/IKAnalyzer.cfg.xml

这里新建一个词库

touch /opt/es/plugins/elasticsearch-analysis-ik-8.15.2/config/ext.dic

编辑扩展词库

vim /opt/es/plugins/elasticsearch-analysis-ik-8.15.2/config/ext.dic

添加分词

铝坨坨

编辑停用词库

vim /opt/es/plugins/elasticsearch-analysis-ik-8.15.2/config/stopword.dic

添加

的

重启ES

docker restart es

测试分词

GET /_analyze
{
  "analyzer": "ik_max_word",
  "text": "重重的铝坨坨"
}

可以看到扩展词库的“铝坨坨”被分词识别出来了，“的”没有被分词
在这里插入图片描述

分词作用

  • 创建倒排索引时对文档分词
  • 用户搜索时对输入的内容分词

IK分词模式

  • ik_smart：智能切分，粗粒度
  • ik_max_word：最细切分，细粒度

DSL 索引操作

  • 仅允许GET, PUT, DELETE, HEAD
  • mapping：对索引库中文档的约束，常见的属性有
    • type：字段数据类型
      • 字符串：text（可分词的文本）、keyword（不分词的精确值，合在一起有意义的词，如国家、品牌）
      • 数值：long、integer、short、byte、double、float
      • 布尔：boolean
      • 日期：date
      • 对象：object
    • index：是否创建倒排索引，默认true
    • analyzer：使用哪种分词器
    • properties：字段的子字段

添加索引库，每次写入操作版本都会+1，如添加（POST）、更新（PUT）

索引库mgr

PUT /mgr
{
  "mappings": {
    "properties": {
      "info": {
        "type": "text",
        "analyzer": "ik_smart"
      },
      "email": {
        "type": "keyword",
        "index": false
      },
      "name": {
        "type": "object",
        "properties": {
          "firstName": {
            "type": "keyword"
          },
          "lastName": {
            "type": "keyword"
          }
        }
      }
    }
  }
}

查询索引库

GET /mgr

更新索引库（索引库禁止修改，因为索引库建立倒排索引后无法修改，只能添加新字段）

PUT /mgr/_mapping
{
  "properties":{
    "age":{
      "type":"integer"
    }
  }
}

删除索引库

DELETE /mgr

DSL文档操作

添加文档

索引库mgr/文档/文档id

POST /mgr/_doc/1
{
  "info": "铝坨坨键盘",
  "email": "[email protected]",
  "name": {
    "firstName": "C",
    "lastName": "I"
  }
}

查询文档

GET /mgr/_doc/1

更新文档

全量更新，删除旧文档，添加新文档

如果文档id不存在则与添加文档功能相同

PUT /mgr/_doc/1
{
  "info": "铝坨坨键盘",
  "email": "[email protected]",
  "name": {
    "firstName": "C",
    "lastName": "I"
  }
}

增量更新（局部更新）

指定_update，指定文档doc

POST /mgr/_update/1
{
  "doc": {
    "email": "[email protected]"
  }
}

删除文档

DELETE /mgr/_doc/1

Rust客户端操作Elasticsearch

添加Cargo.toml

elasticsearch = "8.15.0-alpha.1"
# 序列化和反序列化数据
serde = { version = "1.0.127", features = ["derive"] }
# 序列化JSON
serde_json = "1.0.128"
tokio = { version = "1", features = ["full"] }
# 异步锁
once_cell = "1.20.2"

添加环境变量.env

# 指定当前配置文件
RUN_MODE=development

添加配置settings\development.toml

debug = true
# 指定开发环境配置
profile = "development"
[es]
host = "127.0.0.1"

获取配置config\es.rs

use serde::Deserialize;
#[derive(Debug, Deserialize, Clone)]
pub struct EsConfig {
    host: String,
    port: u16,
}
impl EsConfig {
    // 获取redis连接地址
    pub fn get_url(&self) -> String {
        format!("http://{host}:{port}", host = self.host, port = self.port)
    }
}

将配置存放到AppConfig

#[derive(Debug, Deserialize, Clone)]
pub struct AppConfig {
    pub es:EsConfig,
}
impl AppConfig {
    pub fn read(env_src: Environment) -> Result<Self, config::ConfigError> {
        // 获取配置文件目录
        let config_dir = get_settings_dir()?;
        info!("config_dir: {:#?}", config_dir);
        // 获取配置文件环境
        let run_mode = std::env::var("RUN_MODE")
            .map(|env| Profile::from_str(&env).map_err(|e| ConfigError::Message(e.to_string())))
            .unwrap_or_else(|_e| Ok(Profile::Dev))?;
        // 当前配置文件名
        let profile_filename = format!("{run_mode}.toml");
        // 获取配置
        let config = config::Config::builder()
            // 添加默认配置
            .add_source(config::File::from(config_dir.join("default.toml")))
            // 添加自定义前缀配置
            .add_source(config::File::from(config_dir.join(profile_filename)))
            // 添加环境变量
            .add_source(env_src)
            .build()?;
        info!("Successfully read config profile: {run_mode}.");
        // 反序列化
        config.try_deserialize()
    }
}
// 获取配置文件目录
pub fn get_settings_dir() -> Result<std::path::PathBuf, ConfigError> {
    Ok(get_project_root()
        .map_err(|e| ConfigError::Message(e.to_string()))?
        .join("settings"))
}
#[cfg(test)]
mod tests {
    use crate::config::profile::Profile;

    use self::env::get_env_source;

    pub use super::*;
    #[test]
    pub fn test_profile_to_string() {
        // 设置dev模式
        let profile: Profile = Profile::try_from("development").unwrap();
        println!("profile: {:#?}", profile);
        assert_eq!(profile, Profile::Dev)
    }
    #[test]
    pub fn test_read_app_config_prefix() {
        // 读取配置
        let config = AppConfig::read(get_env_source("APP")).unwrap();
        println!("config: {:#?}", config);
    }
}

将配置存放到全局constant\mod.rs

// 环境变量前缀
pub const ENV_PREFIX: &str = "APP";
// 配置
pub static CONFIG: Lazy<crate::config::AppConfig> = Lazy::new(||
    crate::config::AppConfig::read(get_env_source(ENV_PREFIX)).unwrap()
);

加载配置文件client\builder.rs

use crate::config::AppConfig;
// 传输配置文件到客户端
pub trait ClientBuilder: Sized {
    fn build_from_config(config: &AppConfig) -> Result<Self,InfraError>;
}

Es客户端client\es.rs

InfraError为自定义错误，请修改为你想要的错误，如标准库错误

// 类型别名
pub type EsClient = Arc<Elasticsearch>;
// 加载配置文件
pub trait EsClientExt: Sized {
    fn build_from_config(config: &AppConfig) -> impl Future<Output = Result<Self, InfraError>>;
}

impl EsClientExt for EsClient {
    async fn build_from_config(config: &AppConfig) -> Result<Self, InfraError> {
        // 1、使用single_node方式创建client
        // let transport = Transport::single_node(&config.es.get_url()).unwrap();
        // let client = Elasticsearch::new(transport);
        // Ok(Arc::new(client))

        // 2、使用builder方式创建client，可以添加多个url
        let url = config.es.get_url();
        let url_parsed = url
            .parse::<elasticsearch::http::Url>()
            .map_err(|_| InfraError::OtherError("url err".to_string()))?;
        let conn_pool = SingleNodeConnectionPool::new(url_parsed);
        let transport = TransportBuilder::new(conn_pool)
            .disable_proxy()
            .build()
            .map_err(|_| InfraError::OtherError("transport err".to_string()))?;
        let client = Elasticsearch::new(transport);
        Ok(Arc::new(client))
    }
}

测试client\es.rs，所有请求在body()中定义DSL语句，通过send()发送

#[cfg(test)]
mod tests {
    use elasticsearch::{ cat::CatIndicesParts, DeleteParts, IndexParts, UpdateParts };
    use serde_json::json;
    use super::*;
    use crate::constant::CONFIG;
    #[tokio::test]
    async fn test_add_document() {
        let client_result = EsClient::build_from_config(&CONFIG).await;
        assert!(client_result.is_ok());
        let client = client_result.unwrap();

        let response = client
            .index(IndexParts::IndexId("mgr", "1"))
            .body(
                json!({
                "id": 1,
                "user": "cci",
                "post_date": "2024-01-15T00:00:00Z",
                "message": "Trying out Elasticsearch, so far so good?"
            })
            )
            .send().await;

        assert!(response.is_ok());
        let response = response.unwrap();
        assert!(response.status_code().is_success());
    }

    #[tokio::test]
    async fn test_get_indices() {
        let client_result = EsClient::build_from_config(&CONFIG).await;
        assert!(client_result.is_ok());
        let client = client_result.unwrap();

        let get_index_response = client
            .cat()
            .indices(CatIndicesParts::Index(&["*"]))
            .send().await;

        assert!(get_index_response.is_ok());
    }

    #[tokio::test]
    async fn test_update_document() {
        let client_result = EsClient::build_from_config(&CONFIG).await;
        assert!(client_result.is_ok());
        let client = client_result.unwrap();

        let update_response = client
            .update(UpdateParts::IndexId("mgr", "1"))
            .body(
                json!({
                "doc": {
                    "message": "Updated message"
                }
            })
            )
            .send().await;

        assert!(update_response.is_ok());
        let update_response = update_response.unwrap();
        assert!(update_response.status_code().is_success());
    }

    #[tokio::test]
    async fn test_delete_document() {
        let client_result = EsClient::build_from_config(&CONFIG).await;
        assert!(client_result.is_ok());
        let client = client_result.unwrap();

        let delete_response = client.delete(DeleteParts::IndexId("mgr", "1")).send().await;

        assert!(delete_response.is_ok());
        let delete_response = delete_response.unwrap();
        assert!(delete_response.status_code().is_success());
    }
}

使用流程

        // 1、创建client
        let client_result = EsClient::build_from_config(&CONFIG).await;
        assert!(client_result.is_ok());
        let client = client_result.unwrap();
        // 2、定义DSL语句
        let mut body: Vec<JsonBody<_>> = Vec::with_capacity(4);
        // 添加文档
        body.push(json!({"index": {"_id": "1"}}).into());
        body.push(
            json!({
    "id": 1,
    "user": "kimchy",
    "post_date": "2009-11-15T00:00:00Z",
    "message": "Trying out Elasticsearch, so far so good?"
}).into()
        );
        // 添加文档
        body.push(json!({"index": {"_id": "2"}}).into());
        body.push(
            json!({
    "id": 2,
    "user": "forloop",
    "post_date": "2020-01-08T00:00:00Z",
    "message": "Bulk indexing with the rust client, yeah!"
}).into()
        );
        // 3、发送请求
        let response = client.bulk(BulkParts::Index("mgr")).body(body).send().await.unwrap();

项目地址：https://github.com/VCCICCV/MGR

分析数据结构

mapping要考虑的问题：字段名、数据类型、是否参与搜索（建立倒排索引"index":false，默认true）、是否分词（参与搜索的字段，text分词，keyword、数据类型不分词）、分词器

  • 地理坐标：
    • geo_point：由经度（longitude）和纬度（latitude）确定的一个点，如[ 13.400544, 52.530286 ]
    • geo_shape：由多个geo_point组成的几何图形，如一条线[[13.0, 53.0], [14.0, 52.0]]
  • copy_to：将多个字段组合为一个字段进行索引

Rust客户端操作索引库

生产环境不要使用unwrap()

这里演示在请求正文中操作，使用send()

Transport支持的方法Method：

  • Get：获取资源
  • Put：创建或更新资源（全量更新）
  • Post：创建或更新资源（部分更新）
  • Delete：删除资源
  • Head：获取头信息

send()请求正文需要包含的参数：

  • method：必须
  • path：必须
  • headers：必须
  • query_string：可选
  • body：可选
  • timeout：可选

添加索引库

    #[tokio::test]
    async fn test_create_index() {
        // 1、创建client
        let client_result = EsClient::build_from_config(&CONFIG).await;
        assert!(client_result.is_ok());
        let client = client_result.unwrap();
        // 2、定义DSL语句
        let index_name = "mgr";
        let index_definition =
            json!({
            "mappings":{
                "properties":{
                    "age":{
                        "type":"integer"
                    }
                }
            }
        });
        let body = Some(serde_json::to_vec(&index_definition).unwrap());
        let path = format!("/{}", index_name);
        let headers = HeaderMap::new();
        let query_string = None;
        let timeout = None;
        let method = Method::Put;
        // 3、发送请求
        let response = client.send::<Vec<u8>, ()>(
            method,
            &path,
            headers,
            query_string,
            body,
            timeout
        ).await;
        assert!(response.is_ok());
        let response = response.unwrap();
        assert_eq!(response.status_code().is_success(), true);
    }

你也可以将其简化

    #[tokio::test]
    async fn test_create_index() {
        // 1、创建client
        let client = EsClient::build_from_config(&CONFIG).await.unwrap();
        // 2、定义DSL
        let index_definition =
            json!({
            "mappings":{
                "properties":{
                    "age":{
                        "type":"integer"
                    }
                }
            }
        });
        // 3、发送请求
        let response = client.send::<Vec<u8>, ()>(
            Method::Put,
            format!("/mgr").as_str(),
            HeaderMap::new(),
            None,
            Some(index_definition.to_string().as_bytes().to_vec()),
            None
        ).await;
        assert!(response.is_ok());
        let response = response.unwrap();
        assert_eq!(response.status_code().is_success(), true);
    }

查询索引库是否存在

    #[tokio::test]
    async fn test_query_index() {
        // 1、创建 client
        let client = EsClient::build_from_config(&CONFIG).await.unwrap();
        // 2、定义查询 DSL 语句
        let query = json!({
        "query": {
            "match_all": {}
        }
    });
        // 3、发送请求
        let response = client.send::<Vec<u8>, ()>(
            Method::Get,
            format!("/mgr/_search").as_str(),
            HeaderMap::new(),
            None,
            Some(query.to_string().as_bytes().to_vec()),
            None
        ).await;
        assert!(response.is_ok());
        let response = response.unwrap();
        println!("{:?}", response);
        assert_eq!(response.status_code().is_success(), true);
    }

也可以不定义DSL查询

    #[tokio::test]
    async fn test_query_index2() {
        // 1、创建 client
        let client = EsClient::build_from_config(&CONFIG).await.unwrap();
        // 2、发送请求
        let response = client.send::<Vec<u8>, ()>(
            Method::Get,
            format!("/mgr").as_str(),
            HeaderMap::new(),
            None,
            None,
            None
        ).await;
        assert!(response.is_ok());
        let response = response.unwrap();
        println!("{:?}", response);
        assert_eq!(response.status_code().is_success(), true);
    }

更新索引库

    #[tokio::test]
    async fn test_update_index() {
        // 1、创建 client
        let client = EsClient::build_from_config(&CONFIG).await.unwrap();
        // 2、定义查询 DSL 语句
        let update_content = json!({
            "properties":{
                "age":{
                "type":"integer"
                }
            }
    });
        // 3、发送请求
        let response = client.send::<Vec<u8>, ()>(
            Method::Put,
            format!("/mgr/_mapping").as_str(),
            HeaderMap::new(),
            None,
            Some(update_content.to_string().as_bytes().to_vec()),
            None
        ).await;
        assert!(response.is_ok());
        let response = response.unwrap();
        println!("{:?}", response);
        assert_eq!(response.status_code().is_success(), true);
    }

删除索引库

    #[tokio::test]
    async fn test_delete_index() {
        // 1、创建 client
        let client = EsClient::build_from_config(&CONFIG).await.unwrap();
        // 2、发送请求
        let response = client.send::<(), ()>(
            Method::Delete,
            format!("/mgr").as_str(),
            HeaderMap::new(),
            None,
            None,
            None
        ).await;
        assert!(response.is_ok());
        let response = response.unwrap();
        assert_eq!(response.status_code().is_success(), true);
    }

Rust客户端操作文档

添加文档

    #[tokio::test]
    async fn test_create_doc() {
        // 1、创建 client
        let client = EsClient::build_from_config(&CONFIG).await.unwrap();
        // 2、定义查询 DSL 语句
        let doc_content =
            json!({
            "id": "1",
            "user": "kimchy",
            "post_date": "2009-11-15T00:00:00Z",
            "message": "Trying out Elasticsearch, so far so good?"
        });
        // 3、发送请求
        let response = client.send::<Vec<u8>, ()>(
            Method::Post,
            format!("/mgr/_doc/1").as_str(),
            HeaderMap::new(),
            None,
            Some(doc_content.to_string().as_bytes().to_vec()),
            None
        ).await;
        assert!(response.is_ok());
        let response = response.unwrap();
        println!("{:?}", response);
        assert_eq!(response.status_code().is_success(), true);
    }

查询文档是否存在

    #[tokio::test]
    async fn test_get_doc() {
        // 1、创建 client
        let client = EsClient::build_from_config(&CONFIG).await.unwrap();
        // 2、发送请求
        let response = client.send::<Vec<u8>, ()>(
            Method::Get,
            format!("/mgr/_doc/1").as_str(),
            HeaderMap::new(),
            None,
            None,
            None
        ).await;
        assert!(response.is_ok());
        let response = response.unwrap();
        println!("{:?}", response);
        assert_eq!(response.status_code().is_success(), true);
    }

更新文档

    #[tokio::test]
    async fn test_update_doc() {
        // 1、创建 client
        let client = EsClient::build_from_config(&CONFIG).await.unwrap();
        // 2、定义查询 DSL 语句
        let doc_content =
            json!({
            "doc": {
                "message": "Updated message"
            }
        });
        // 3、发送请求
        let response = client.send::<Vec<u8>, ()>(
            Method::Post,
            format!("/mgr/_update/1").as_str(),
            HeaderMap::new(),
            None,
            Some(doc_content.to_string().as_bytes().to_vec()),
            None
        ).await;
        assert!(response.is_ok());
        let response = response.unwrap();
        println!("{:?}", response);
        assert_eq!(response.status_code().is_success(), true);
    }

删除文档

    #[tokio::test]
    async fn test_delete_doc() {
        // 1、创建 client
        let client = EsClient::build_from_config(&CONFIG).await.unwrap();
        // 2、发送请求
        let response = client.send::<Vec<u8>, ()>(
            Method::Delete,
            format!("/mgr/_doc/1").as_str(),
            HeaderMap::new(),
            None,
            None,
            None
        ).await;
        assert!(response.is_ok());
        let response = response.unwrap();
        println!("{:?}", response);
        assert_eq!(response.status_code().is_success(), true);
    }

批量添加文档

    #[tokio::test]
    async fn test_bulk_add_to_mgr() {
        // 1、创建client
        let client_result = EsClient::build_from_config(&CONFIG).await;
        assert!(client_result.is_ok());
        let client = client_result.unwrap();
        // 2、定义DSL语句
        let mut body: Vec<JsonBody<_>> = Vec::with_capacity(4);
        // 添加第一个操作和文档
        body.push(json!({"index": {"_id": "1"}}).into());
        body.push(
            json!({
    "id": 1,
    "user": "kimchy",
    "post_date": "2009-11-15T00:00:00Z",
    "message": "Trying out Elasticsearch, so far so good?"
}).into()
        );

        // 添加第二个操作和文档
        body.push(json!({"index": {"_id": "2"}}).into());
        body.push(
            json!({
    "id": 2,
    "user": "forloop",
    "post_date": "2020-01-08T00:00:00Z",
    "message": "Bulk indexing with the rust client, yeah!"
}).into()
        );
        // 3、发送请求
        let response = client.bulk(BulkParts::Index("mgr")).body(body).send().await.unwrap();
        assert!(response.status_code().is_success());
    }

Rust客户端操作搜索

这里演示在请求体body中进行API调用

  • 查询所有：查出所有数据
  • 全文检索查询（full text）：利用分词器对内容分词，从倒排索引库中查询
    • match_query
    • multi_match_query
  • 精确查询：根据精确值查询，如integer、keyword、日期
    • id
    • range：根据值的范围查询
    • term：根据词条精确值查询
  • 地理坐标查询（geo）：根据经纬度查询
    • geo_distance：查询geo_point指定距离范围内的所有文档
    • geo_bounding_box：查询geo_point值落在某个矩形范围内的所有文档
  • 复合查询（compound）：将上述条件组合起来

查询所有

默认10条

    #[tokio::test]
    async fn test_search_match_all() {
        // 1、创建 client
        let client = EsClient::build_from_config(&CONFIG).await.unwrap();
        // 2. 执行搜索
        let response = client
            .search(SearchParts::Index(&["mgr"]))
            .from(0)
            .size(5)
            .body(
                json!({
            "query": {
                "match_all": {
                }
            }
        })
            )
            .send().await
            .unwrap();
        // 3. 解析响应
        let response_body = response.json::<Value>().await.unwrap();
        // 搜索耗时
        let took = response_body["took"].as_i64().unwrap();
        println!("took: {}ms", took);
        // 搜索结果
        for hit in response_body["hits"]["hits"].as_array().unwrap() {
            println!("{:?}", hit["_source"]);
        }
    }

等价于

GET /mgr/_search
{
  "query": {
    "match_all": {}
  }
}

全文搜索

message为文档中的字段

    #[tokio::test]
    async fn test_search_match() {
        // 1、创建 client
        let client = EsClient::build_from_config(&CONFIG).await.unwrap();
        // 2. 执行搜索
        let response = client
            .search(SearchParts::Index(&["mgr"]))
            .from(0)
            .size(5)
            .body(
                json!({
            "query": {
                "match": {
                    "message": "good"
                }
            }
        })
            )
            .send().await
            .unwrap();
        // 3. 解析响应
        let response_body = response.json::<Value>().await.unwrap();
        // 搜索耗时
        let took = response_body["took"].as_i64().unwrap();
        println!("took: {}ms", took);
        // 搜索结果
        for hit in response_body["hits"]["hits"].as_array().unwrap() {
            println!("{:?}", hit["_source"]);
        }
    }

相当于

GET /mgr/_search
{
  "query": {
    "match": {
      "message": "good"
    }
  }
}

多字段查询

多字段查询效率低，一般在创建时使用copy_to到一个字段中

    #[tokio::test]
    async fn test_search_multi_match() {
        // 1、创建 client
        let client = EsClient::build_from_config(&CONFIG).await.unwrap();
        // 2. 执行搜索
        let response = client
            .search(SearchParts::Index(&["mgr"]))
            .from(0)
            .size(5)
            .body(
                json!({
            "query": {
                "multi_match": {
                    "query": "good",
                    "fields": [
                        "message",
                        "user"
                        ]
                    }
                }
            })
            )
            .send().await
            .unwrap();
        // 3. 解析响应
        let response_body = response.json::<Value>().await.unwrap();
        // 搜索耗时
        let took = response_body["took"].as_i64().unwrap();
        println!("took: {}ms", took);
        // 搜索结果
        for hit in response_body["hits"]["hits"].as_array().unwrap() {
            println!("{:?}", hit["_source"]);
        }
    }

相当于

GET /mgr/_search
{
  "query": {
    "multi_match": {
      "query": "good",
      "fields": [
        "message",
        "user"
      ]
    }
  }
}

根据范围查询（range）

gte大于等于，lte小于等于；gt大于lt小于

    #[tokio::test]
    async fn test_search_range() {
        // 1、创建 client
        let client = EsClient::build_from_config(&CONFIG).await.unwrap();
        // 2. 执行搜索
        let response = client
            .search(SearchParts::Index(&["mgr"]))
            .from(0)
            .size(5)
            .body(
                json!({
            "query": {
                "range": {
                    "id": {
                        "gte": 1,
                        "lte": 1
                        }
                    }
                }
            })
            )
            .send().await
            .unwrap();
        // 3. 解析响应
        let response_body = response.json::<Value>().await.unwrap();
        // 搜索耗时
        let took = response_body["took"].as_i64().unwrap();
        println!("took: {}ms", took);
        // 搜索结果
        for hit in response_body["hits"]["hits"].as_array().unwrap() {
            println!("{:?}", hit["_source"]);
        }
    }

相当于

GET /mgr/_search
{
  "query": {
    "range": {
      "id": {
        "gte": 1,
        "lte": 1
      }
    }
  }
}

根据词条精确查询（term）

    #[tokio::test]
    async fn test_search_term() {
        // 1、创建 client
        let client = EsClient::build_from_config(&CONFIG).await.unwrap();
        // 2. 执行搜索
        let response = client
            .search(SearchParts::Index(&["mgr"]))
            .from(0)
            .size(5)
            .body(
                json!({
            "query": {
                "term": {
                    "user": "kimchy"
                    }
                }
            })
            )
            .send().await
            .unwrap();
        // 3. 解析响应
        let response_body = response.json::<Value>().await.unwrap();
        // 搜索耗时
        let took = response_body["took"].as_i64().unwrap();
        println!("took: {}ms", took);
        // 搜索结果
        for hit in response_body["hits"]["hits"].as_array().unwrap() {
            println!("{:?}", hit["_source"]);
        }
    }

相当于

GET /mgr/_search
{
  "query": {
    "term": {
      "user": "kimchy"
    }
  }
}

根据地理距离查询

GET /mgr/_search
{
  "query": {
    "geo_distance": {
      "distance": "100km",
      "location": "31.04, 45.12"
    }
  }
}

根据指定矩形范围查询

左上经纬度与右下经纬度
geo为文档中的字段

GET /mgr/_search
{
  "query": {
    "geo_bounding_box": {
      "geo": {
        "top_left": {
          "lon": 124.45,
          "lat": 32.11
        },
        "bottom_right": {
          "lon": 125.12,
          "lat": 30.21
        }
      }
    }
  }
}

复合查询

查询时文档会对搜索词条的关联度打分_score，返回结果时按照降序排列

关联度计算方法

  • TF-IDF算法（ES5.0之前）

TF（词条频率）= 词条出现次数/文档中词条总数

IDF（逆文档频率）=log(文档总数/包含词条的文档总数)

score = ∑(

标签：unwrap,client,let,整合,test,Elasticsearch,config,response,Rust
From： https://blog.csdn.net/weixin_62799021/article/details/143243497

相关文章

  • 刚刚，Stable Diffusion 2024升级，最强Ai绘画整合包、部署教程（解压即用）
    2024Ai技术大爆发的元年目前两款Ai神器大火一款是大名鼎鼎的ChatGPT另外一款—StableDiffusion堪称全球最强Ai绘画工具StableDiffusionAi绘画2024版本更新啦！从4.8.7更新至**4.9版本！**更新优化和大模型增加，无需安装，解压即用sd整合包获取方式Stablediffusion......
  • elasticsearch使用
    1、选择1、ElasticsearchRestTemplate是spring对官方HighLevelRESTClient的封装。2、ElasticSearch8.x弃用了HighLevelRESTClient，移除了JavaTransportClient，推荐使用ElasticsearchJavaAPI（后续使用8的建议使用ElasticsearchJavaAPI）2、ElasticsearchRestTemp......
  • Springboot整合Swagger3
    1、pom.xml依赖<!--Swagger3依赖--><dependency><groupId>io.springfox</groupId><artifactId>springfox-boot-starter</artifactId><version>${swagger.version}</version>......
  • SpringBoot整合Mybatis
    1、项目结构 2、pom.xml依赖<?xmlversion="1.0"encoding="UTF-8"?><projectxmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation=&q......
  • C语言和Rust在安全性特性上的区别
    #C语言和Rust在安全性特性上的区别在探讨C语言和Rust在安全性特性上的区别时，我们可以明确地指出几个核心观点：Rust提供了内存安全保证、并发安全、以及错误处理机制，这些特性在编译时就能够避免许多常见的错误类型，显著提高了软件的安全性和可靠性。其中，内存安全保证是Rust最为突......
  • ASP.Net Core 8 Web API整合Swagger UI并进行模块分组吃屎般瞬间记录
    一、开发环境开发工具：VisualStudio2022工程模板：ASP.NetCore8WebAPI工程(官方标准的).Net环境：.NetCore8.0NuGet依赖：Swashbuckle.AspNetCore6.9.0(UI用的默认的UI界面,可以自由选择其他的UI界面)二、基本概述参考了网上很多大佬的帖子，实现基本就两种：1、用自定义At......
  • Rust 基础语法与常用特性
    Rust跨界：全面掌握跨平台应用开发第一章：快速上手Rust1.2基础语法与常用特性1.2.1数据类型与控制流数据类型Rust提供了丰富的内置数据类型，主要分为标量类型和复合类型。标量类型标量类型表示单一的值，Rust中的标量类型包括：整数类型：Rust提供了多种整数类型，分为有符......
  • Rust 跨平台应用开发第一章：快速上手 Rust——实用示例
    1.3实用示例在这一节中，我们将通过一系列实用的示例来帮助您更好地理解Rust的特性，并展示如何在实际项目中使用这些特性。示例将涵盖文件操作、网络请求、并发编程、命令行工具以及使用Cargo管理依赖等多个方面。1.3.1文件操作示例Rust提供了强大的标准库来进行文件操......
  • ElasticSearch安装与使用
    一、ElasticSearch的安装下载ElasticSearch安装包可以从ElasticSearch的官方网站下载相应版本的安装包。或者，在Linux系统中，可以使用wget命令下载，例如：wgethttps://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.x.x-linux-x86_64.tar.gz（请将7.x.x替换为具......
  • 从0到1，用Rust轻松制作电子书
    我之前简单提到过用Rust做电子书，今天分享下如何用Rust做电子书。制作电子书其实用途广泛，不仅可以用于技术文档（对技术人来说非常方便），也可以制作用户手册、笔记、教程等，还可以应用于文学创作。如果你有想法写小说或者做知识付费，电子书也是一个不错的切入点。特别是知识付费领......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99