Rust client
출처: 문서
본문
Rust 클라이언트
OpenSearch Rust 클라이언트를 사용하면 Rust 애플리케이션을 OpenSearch 클러스터의 데이터와 연결할 수 있어요. 클라이언트의 전체 API 문서와 추가 예제는 OpenSearch docs.rs 문서를 참고하세요.
이 시작 가이드에서는 OpenSearch에 연결하고, 문서를 인덱싱하고, 쿼리를 실행하는 방법을 다뤄요. 클라이언트 소스 코드는 opensearch-rs 저장소를 참고하세요.
Rust 클라이언트 설치
새 프로젝트를 시작한다면 Cargo.toml에 opensearch 크레이트를 추가하세요:
[dependencies]
opensearch = "2.4.0"
{% include copy.html %}
또한 타입을 JSON으로 직렬화하고 JSON 응답을 역직렬화하는 데 도움이 되는 다음 serde 의존성을 추가할 수도 있어요. derive 기능을 사용하면 직접 만든 구조체에 대해 Serialize와 Deserialize를 derive할 수 있어요:
serde = { version = "~1", features = ["derive"] }
serde_json = "~1"
{% include copy.html %}
Rust 클라이언트는 HTTP 요청에 고수준 HTTP 클라이언트 라이브러리인 reqwest를 사용하고, reqwest는 비동기 요청을 지원하기 위해 tokio 플랫폼을 사용해요. 비동기 함수를 사용할 계획이라면 Cargo.toml에 tokio 의존성을 추가해야 해요:
tokio = { version = "1", features = ["full"] }
{% include copy.html %}
전체 Cargo.toml 파일은 샘플 프로그램 섹션을 참고하세요.
Rust 클라이언트 API를 사용하려면 필요한 모듈, 구조체, 열거형을 임포트하세요:
use opensearch::OpenSearch;
{% include copy.html %}
샘플 데이터
이 페이지의 예제에서는 문서를 나타내기 위해 Student 구조체를 사용해요. #[serde(rename_all = "camelCase")] 어트리뷰트는 구조체 필드를 firstName, lastName, gpa, gradDate JSON 필드로 직렬화해요:
use serde::{Deserialize, Serialize};
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
struct Student {
first_name: String,
last_name: String,
gpa: f64,
grad_date: String,
}
impl Student {
fn new(first_name: &str, last_name: &str, gpa: f64, grad_date: &str) -> Self {
Student {
first_name: first_name.to_string(),
last_name: last_name.to_string(),
gpa,
grad_date: grad_date.to_string(),
}
}
}
{% include copy.html %}
OpenSearch에 연결
기본 OpenSearch 호스트에 연결하려면 http://localhost:9200 주소로 OpenSearch에 접속하는 기본 클라이언트 객체를 생성하세요:
let client = OpenSearch::default();
{% include copy.html %}
이 페이지의 나머지 연결 예제에는 다음 임포트가 필요해요:
use opensearch::{
http::transport::{SingleNodeConnectionPool, Transport, TransportBuilder},
http::Url,
OpenSearch,
};
{% include copy.html %}
다른 주소에서 실행 중인 OpenSearch 호스트에 연결하려면 지정한 주소로 클라이언트를 생성하세요:
let transport = Transport::single_node("http://localhost:9200")?;
let client = OpenSearch::new(transport);
{% include copy.html %}
또는 TransportBuilder 구조체를 만들어 URL을 커스터마이즈하고 커넥션 풀을 사용할 수 있어요. 이 구조체를 OpenSearch::new에 전달하면 클라이언트의 새 인스턴스가 생성돼요:
let url = Url::parse("http://localhost:9200")?;
let conn_pool = SingleNodeConnectionPool::new(url);
let transport = TransportBuilder::new(conn_pool).disable_proxy().build()?;
let client = OpenSearch::new(transport);
{% include copy.html %}
Security 플러그인이 활성화된 클러스터에 연결하려면 HTTPS를 사용하고 기본 인증 자격 증명을 제공하세요. 다음 예제에서는 클라이언트가 자체 서명된 데모 인증서를 수락하도록 인증서 검증도 비활성화해요. opensearch::auth에서 Credentials를, opensearch::cert에서 CertificateValidation을 임포트하세요:
let url = Url::parse("https://localhost:9200")?;
let conn_pool = SingleNodeConnectionPool::new(url);
let transport = TransportBuilder::new(conn_pool)
// Only for demo purposes. Don't specify your credentials in code.
.auth(Credentials::Basic(
"admin".to_string(),
"<custom-admin-password>".to_string(),
))
// Only for demo purposes. Disables certificate validation for self-signed certificates.
.cert_validation(CertificateValidation::None)
.build()?;
let client = OpenSearch::new(transport);
{% include copy.html %}
Amazon OpenSearch Service에 연결
AWS Signature Version 4로 요청에 서명하려면 opensearch 크레이트의 aws-auth 기능을 활성화하고 Cargo.toml에 aws-config 의존성을 추가하세요:
opensearch = { version = "2.4.0", features = ["aws-auth"] }
aws-config = "1"
{% include copy.html %}
그다음 AWS 구성 타입을 임포트하세요:
use aws_config::{meta::region::RegionProviderChain, BehaviorVersion};
{% include copy.html %}
다음 예제에서는 엔드포인트를 여러분의 도메인 엔드포인트로 바꾸세요. 도메인 엔드포인트는 Amazon OpenSearch Service 콘솔의 도메인 세부 정보 페이지에 표시돼요.
다음 예제는 Amazon OpenSearch Service에 연결하는 방법을 보여줘요:
let url = Url::parse("https://search-<domain-name>-<id>.us-east-1.es.amazonaws.com")?;
let service_name = "es";
let conn_pool = SingleNodeConnectionPool::new(url);
let region_provider = RegionProviderChain::default_provider().or_else("us-east-1");
let aws_config = aws_config::defaults(BehaviorVersion::latest())
.region(region_provider)
.load()
.await;
let transport = TransportBuilder::new(conn_pool)
.auth(aws_config.try_into()?)
.service_name(service_name)
.build()?;
let client = OpenSearch::new(transport);
{% include copy.html %}
Amazon OpenSearch Serverless에 연결
다음 예제에서는 엔드포인트를 여러분의 컬렉션 엔드포인트로 바꾸세요. 컬렉션 엔드포인트는 Amazon OpenSearch Service 콘솔의 컬렉션 세부 정보 페이지에 표시돼요.
Amazon OpenSearch Serverless에 연결하려면 Amazon OpenSearch Service에 연결과 동일한 aws-auth 기능, aws-config 의존성, 임포트가 필요해요. 다음 예제는 Amazon OpenSearch Serverless에 연결하는 방법을 보여줘요:
let url = Url::parse("https://<collection-id>.us-east-1.aoss.amazonaws.com")?;
let service_name = "aoss";
let conn_pool = SingleNodeConnectionPool::new(url);
let region_provider = RegionProviderChain::default_provider().or_else("us-east-1");
let aws_config = aws_config::defaults(BehaviorVersion::latest())
.region(region_provider)
.load()
.await;
let transport = TransportBuilder::new(conn_pool)
.auth(aws_config.try_into()?)
.service_name(service_name)
.build()?;
let client = OpenSearch::new(transport);
{% include copy.html %}
Amazon OpenSearch Serverless는 OpenSearch API 작업 중 일부만 지원하며, 이 페이지의 예제에서 사용하는 refresh 파라미터는 지원하지 않아요. 자세한 내용은 Amazon OpenSearch Serverless에서 지원되는 작업 및 플러그인을 참고하세요.
{: .note}
인덱스 생성
다음 예제는 프라이머리 샤드 1개와 레플리카 1개를 가진 인덱스를 생성해요. gradDate 필드를 yyyy-MM-dd 형식의 date로 명시적으로 매핑해요. 문서를 인덱싱할 때 나머지 문서 필드는 OpenSearch가 동적으로 매핑해요:
let index = "students";
let response = client
.indices()
.create(IndicesCreateParts::Index(index))
.body(json!({
"settings": {
"index": {
"number_of_shards": 1,
"number_of_replicas": 1
}
},
"mappings": {
"properties": {
"gradDate": { "type": "date", "format": "yyyy-MM-dd" }
}
}
}))
.send()
.await?;
{% include copy.html %}
문서 인덱싱
클라이언트의 index 함수를 사용하면 문서를 OpenSearch에 인덱싱할 수 있어요. refresh(Refresh::True) 호출을 하면 문서를 바로 검색할 수 있게 돼요. Refresh는 opensearch::params 모듈에 정의되어 있어요:
let student = Student::new("John", "Doe", 3.89, "2022-05-15");
let response = client
.index(IndexParts::IndexId(index, "1"))
.body(student)
.refresh(Refresh::True)
.send()
.await?;
{% include copy.html %}
벌크 작업 수행
클라이언트의 bulk 함수를 사용하면 여러 작업을 한꺼번에 수행할 수 있어요. 먼저 Bulk API 호출의 JSON 본문을 만든 다음 bulk 함수에 전달하세요:
let body: Vec<JsonBody<Value>> = vec![
json!({"index": {"_id": "2"}}).into(),
serde_json::to_value(Student::new("Paulo", "Santos", 3.93, "2021-05-20"))?.into(),
json!({"index": {"_id": "3"}}).into(),
serde_json::to_value(Student::new("Shirley", "Rodriguez", 3.91, "2019-05-10"))?.into(),
];
let response = client
.bulk(BulkParts::Index(index))
.body(body)
.refresh(Refresh::True)
.send()
.await?;
{% include copy.html %}
문서 검색
인덱스에 있는 모든 문서를 검색하려면 쿼리 없이 검색 요청을 보내세요:
let response = client
.search(SearchParts::Index(&[index]))
.send()
.await?;
{% include copy.html %}
그다음 응답 본문을 JSON으로 읽고 hits 배열을 순회하면서 각 _source 문서를 Student로 역직렬화할 수 있어요:
let response_body = response.json::<Value>().await?;
println!("Total hits: {}", response_body["hits"]["total"]["value"]);
for hit in response_body["hits"]["hits"].as_array().unwrap_or(&vec![]) {
let student: Student = serde_json::from_value(hit["_source"].clone())?;
println!(" {}", serde_json::to_string(&student)?);
}
{% include copy.html %}
각 _source 문서는 Student 구조체로 역직렬화되기 때문에 student.first_name처럼 구조체 필드 형태로 문서 필드에 접근할 수 있어요. 각 hit에는 문서 ID가 hit["_id"]에, 문서 자체는 hit["_source"]에 담겨 있어요. 각 문서의 ID와 필드를 출력하려면 다음 코드를 사용하세요:
for hit in response_body["hits"]["hits"].as_array().unwrap_or(&vec![]) {
let id = hit["_id"].as_str().unwrap_or_default();
let student: Student = serde_json::from_value(hit["_source"].clone())?;
println!(
"ID: {}, name: {} {}, GPA: {}, graduation date: {}",
id, student.first_name, student.last_name, student.gpa, student.grad_date
);
}
{% include copy.html %}
2019년에 졸업한 학생을 검색하려면 gradDate 필드에 range 쿼리를 사용하세요:
let response = client
.search(SearchParts::Index(&[index]))
.body(json!({
"query": {
"range": {
"gradDate": {
"gte": "2019-01-01",
"lte": "2019-12-31"
}
}
}
}))
.send()
.await?;
{% include copy.html %}
결과 페이지네이션
결과를 페이지네이션하려면 from과 size 파라미터를 사용하세요. 다음 예제는 학생들을 졸업 날짜 기준으로 정렬하고 결과를 한 번에 2개씩 가져와요. 첫 번째 요청은 결과의 첫 페이지를 반환하고, 두 번째 요청은 그다음 페이지를 반환해요:
let response = client
.search(SearchParts::Index(&[index]))
.from(0)
.size(2)
.sort(&["gradDate:asc"])
.send()
.await?;
let next_page = client
.search(SearchParts::Index(&[index]))
.from(2)
.size(2)
.sort(&["gradDate:asc"])
.send()
.await?;
{% include copy.html %}
from과 size 파라미터는 결과의 앞쪽 페이지를 조회할 때 잘 동작해요. 많은 양의 결과를 페이지네이션하려면 point in time과 search_after를 함께 사용하세요. 자세한 내용은 결과 페이지네이션을 참고하세요.
문서 업데이트
클라이언트의 update 함수를 사용하면 문서를 업데이트할 수 있어요. 다음 예제는 ID가 1인 문서의 gpa 필드를 3.92로 설정해요:
let response = client
.update(UpdateParts::IndexId(index, "1"))
.body(json!({
"doc": {
"gpa": 3.92
}
}))
.send()
.await?;
{% include copy.html %}
업데이트된 문서를 조회하려면 클라이언트의 get 함수를 사용하세요:
let response = client
.get(GetParts::IndexId(index, "1"))
.send()
.await?;
{% include copy.html %}
문서 삭제
클라이언트의 delete 함수를 사용하면 문서를 삭제할 수 있어요:
let response = client
.delete(DeleteParts::IndexId(index, "3"))
.refresh(Refresh::True)
.send()
.await?;
{% include copy.html %}
인덱스 삭제
opensearch::indices::Indices 구조체의 delete 함수를 사용하면 인덱스를 삭제할 수 있어요:
let response = client
.indices()
.delete(IndicesDeleteParts::Index(&[index]))
.send()
.await?;
{% include copy.html %}
샘플 프로그램
이 샘플 프로그램은 앞선 섹션들의 코드를 하나로 합친 거예요. Security 플러그인이 활성화된 클러스터에 연결해요. Security 플러그인이 없는 클러스터에 연결하려면 // Without security 주석이 붙은 줄을 변경하세요.
샘플 프로그램은 Rust 클라이언트 설치 섹션에서 설명한 모든 의존성이 담긴 다음 Cargo.toml 파일을 사용해요:
[package]
name = "os_rust_project"
version = "0.1.0"
edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
opensearch = "2.4.0"
tokio = { version = "1", features = ["full"] }
serde = { version = "~1", features = ["derive"] }
serde_json = "~1"
{% include copy.html %}
이 샘플 프로그램은 테스트 전용이에요. 자체 서명 인증서를 사용하는 클러스터에 연결할 수 있도록 코드에 자격 증명을 지정하고 인증서 검증을 비활성화해요. 프로덕션 환경에서는 자격 증명을 안전한 위치에서 불러오고 클러스터 인증서를 검증하세요. {: .warning}
다음 샘플 프로그램은 클라이언트를 생성하고, 인덱스를 만들고, 문서를 개별적으로 그리고 벌크로 인덱싱하고, 문서를 검색하고, 문서를 업데이트하고, 문서를 삭제한 뒤 인덱스를 삭제해요:
use opensearch::{
auth::Credentials, // Without security, remove this line
cert::CertificateValidation, // Without security, remove this line
http::request::JsonBody,
http::transport::{SingleNodeConnectionPool, TransportBuilder},
http::Url,
indices::{IndicesCreateParts, IndicesDeleteParts},
params::Refresh,
BulkParts, DeleteParts, GetParts, IndexParts, OpenSearch, SearchParts, UpdateParts,
};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
struct Student {
first_name: String,
last_name: String,
gpa: f64,
grad_date: String,
}
impl Student {
fn new(first_name: &str, last_name: &str, gpa: f64, grad_date: &str) -> Self {
Student {
first_name: first_name.to_string(),
last_name: last_name.to_string(),
gpa,
grad_date: grad_date.to_string(),
}
}
}
fn print_students(response_body: &Value) -> Result<(), Box<dyn std::error::Error>> {
for hit in response_body["hits"]["hits"].as_array().unwrap_or(&vec![]) {
let student: Student = serde_json::from_value(hit["_source"].clone())?;
println!(" {}", serde_json::to_string(&student)?);
}
Ok(())
}
fn print_hits(response_body: &Value) -> Result<(), Box<dyn std::error::Error>> {
println!("Total hits: {}", response_body["hits"]["total"]["value"]);
print_students(response_body)
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let url = Url::parse("https://localhost:9200")?; // Without security, use http://localhost:9200
let conn_pool = SingleNodeConnectionPool::new(url);
let transport = TransportBuilder::new(conn_pool)
// Without security, remove this line
.auth(Credentials::Basic("admin".to_string(), "<custom-admin-password>".to_string()))
.cert_validation(CertificateValidation::None) // Without security, remove this line
.build()?;
let client = OpenSearch::new(transport);
// Create the index
let index = "students";
println!("Creating index......");
let response_body = client
.indices()
.create(IndicesCreateParts::Index(index))
.body(json!({
"settings": {
"index": {
"number_of_shards": 1,
"number_of_replicas": 1
}
},
"mappings": {
"properties": {
"gradDate": { "type": "date", "format": "yyyy-MM-dd" }
}
}
}))
.send()
.await?
.json::<Value>()
.await?;
println!("Index created: {}", response_body["index"].as_str().unwrap_or_default());
// Index a document
println!("\nIndexing one student......");
let student = Student::new("John", "Doe", 3.89, "2022-05-15");
let response_body = client
.index(IndexParts::IndexId(index, "1"))
.body(student)
.refresh(Refresh::True)
.send()
.await?
.json::<Value>()
.await?;
println!(
"Result: {}, id: {}, version: {}",
response_body["result"].as_str().unwrap_or_default(),
response_body["_id"].as_str().unwrap_or_default(),
response_body["_version"]
);
// Bulk index documents
println!("\nIndexing many students......");
let body: Vec<JsonBody<Value>> = vec![
json!({"index": {"_id": "2"}}).into(),
serde_json::to_value(Student::new("Paulo", "Santos", 3.93, "2021-05-20"))?.into(),
json!({"index": {"_id": "3"}}).into(),
serde_json::to_value(Student::new("Shirley", "Rodriguez", 3.91, "2019-05-10"))?.into(),
];
let response_body = client
.bulk(BulkParts::Index(index))
.body(body)
.refresh(Refresh::True)
.send()
.await?
.json::<Value>()
.await?;
println!("Errors: {}", response_body["errors"]);
for item in response_body["items"].as_array().unwrap_or(&vec![]) {
println!(
" {} id: {}",
item["index"]["result"].as_str().unwrap_or_default(),
item["index"]["_id"].as_str().unwrap_or_default()
);
}
// Search for all students, two at a time, sorted by graduation date
println!("\nSearching for all students......");
for (page, from) in [(1, 0), (2, 2)] {
let response_body = client
.search(SearchParts::Index(&[index]))
.from(from)
.size(2)
.sort(&["gradDate:asc"])
.send()
.await?
.json::<Value>()
.await?;
if page == 1 {
println!("Total hits: {}", response_body["hits"]["total"]["value"]);
}
println!("Page {}:", page);
print_students(&response_body)?;
}
// Search for students who graduated in 2019
println!("\nSearching for students who graduated in 2019......");
let response_body = client
.search(SearchParts::Index(&[index]))
.body(json!({
"query": {
"range": {
"gradDate": {
"gte": "2019-01-01",
"lte": "2019-12-31"
}
}
}
}))
.send()
.await?
.json::<Value>()
.await?;
print_hits(&response_body)?;
// Update a document
println!("\nUpdating a student's GPA......");
let response_body = client
.update(UpdateParts::IndexId(index, "1"))
.body(json!({
"doc": {
"gpa": 3.92
}
}))
.send()
.await?
.json::<Value>()
.await?;
println!(
"Result: {}, version: {}",
response_body["result"].as_str().unwrap_or_default(),
response_body["_version"]
);
// Get the updated document
let response_body = client
.get(GetParts::IndexId(index, "1"))
.send()
.await?
.json::<Value>()
.await?;
let student: Student = serde_json::from_value(response_body["_source"].clone())?;
println!("Updated document: {}", serde_json::to_string(&student)?);
// Delete a document
println!("\nDeleting a student......");
let response_body = client
.delete(DeleteParts::IndexId(index, "3"))
.refresh(Refresh::True)
.send()
.await?
.json::<Value>()
.await?;
println!("Result: {}", response_body["result"].as_str().unwrap_or_default());
// Delete the index
println!("\nDeleting the index......");
let response_body = client
.indices()
.delete(IndicesDeleteParts::Index(&[index]))
.send()
.await?
.json::<Value>()
.await?;
println!("Acknowledged: {}", response_body["acknowledged"]);
Ok(())
}
{% include copy.html %}
프로그램을 실행하면 다음과 같은 출력이 나와요:
Creating index......
Index created: students
Indexing one student......
Result: created, id: 1, version: 1
Indexing many students......
Errors: false
created id: 2
created id: 3
Searching for all students......
Total hits: 3
Page 1:
{"firstName":"Shirley","lastName":"Rodriguez","gpa":3.91,"gradDate":"2019-05-10"}
{"firstName":"Paulo","lastName":"Santos","gpa":3.93,"gradDate":"2021-05-20"}
Page 2:
{"firstName":"John","lastName":"Doe","gpa":3.89,"gradDate":"2022-05-15"}
Searching for students who graduated in 2019......
Total hits: 1
{"firstName":"Shirley","lastName":"Rodriguez","gpa":3.91,"gradDate":"2019-05-10"}
Updating a student's GPA......
Result: updated, version: 2
Updated document: {"firstName":"John","lastName":"Doe","gpa":3.92,"gradDate":"2022-05-15"}
Deleting a student......
Result: deleted
Deleting the index......
Acknowledged: true
관련 문서
- 클라이언트 사용 예제를 더 보려면
opensearch-rs사용자 가이드를 참고하세요. - 벌크 인덱싱, 검색 같은 특정 작업의 가이드는
opensearch-rs가이드를 참고하세요.