ElasticsearchSQLRetriever

Elasticsearch Document Store에 대해 원시 Elasticsearch SQL 쿼리를 실행하고 원시 JSON 응답을 반환해요.

출처: 문서

본문

파이프라인에서의 일반적인 위치: 단독으로, 또는 메타데이터·집계·구조화 데이터를 가져와야 하는 곳 어디든 사용해요.
필수 초기화 변수: document_store(ElasticsearchDocumentStore 인스턴스)
필수 실행 변수: query(Elasticsearch SQL 쿼리 문자열)
출력 변수: result(Elasticsearch SQL API의 원시 JSON 응답을 담은 딕셔너리)

개요 (Overview)

ElasticsearchSQLRetriever는 ElasticsearchDocumentStore에 대해 Elasticsearch SQL 쿼리를 직접 실행할 수 있게 해줘요. ElasticsearchBM25Retriever나 ElasticsearchEmbeddingRetriever처럼 문서에 대해 쿼리를 매칭하는 대신, SQL 문을 실행하고 Elasticsearch SQL API의 원시 JSON 응답을 반환해요.

이는 런타임에 인덱스에 대한 구조화된 접근이 필요할 때 유용해요. 예를 들어 특정 필드를 가져오거나, 메타데이터로 필터링하거나, 개수·평균 같은 집계를 계산할 때요.

다른 Elasticsearch 리트리버와 달리, 이 컴포넌트는 Document 객체 리스트를 반환하지 않아요. 출력은 단일 result 딕셔너리이며, result["result"]에 Elasticsearch 원시 응답이 담겨요. 일반적인 쿼리의 응답에는 다음이 포함돼요.

result["result"]["columns"]: 반환된 각 열을 설명하는 메타데이터.
result["result"]["rows"]: 데이터 행.

이 컴포넌트는 초기화 시 두 가지 선택적 파라미터를 받아요.

raise_on_failure: True(기본값)이면 SQL API 호출이 실패할 때 예외가 발생해요. False면 오류를 경고로 기록하고 빈 딕셔너리를 반환해요.
fetch_size: 페이지당 가져올 결과 수. 설정하지 않으면 Elasticsearch에 구성된 기본 fetch size를 사용해요.

설치 (Installation)

Elasticsearch를 설치한 뒤 인스턴스를 시작하세요. Haystack은 Elasticsearch 8을 지원해요.

Docker가 설정되어 있다면 Docker 이미지를 받아 실행하는 것을 권장해요.

docker pull docker.elastic.co/elasticsearch/elasticsearch:8.19.7
docker run -p 9200:9200 -e "discovery.type=single-node" -e "ES_JAVA_OPTS=-Xms1024m -Xmx1024m" -e "xpack.security.enabled=false" elasticsearch:8.19.7

대안으로 Elasticsearch 통합 GitHub에서 제공하는 docker-compose.yml을 사용해 Elasticsearch를 실행하는 Docker 컨테이너를 시작할 수 있어요.

docker compose up

실행 중인 Elasticsearch 인스턴스가 있으면 elasticsearch-haystack 통합을 설치하세요.

pip install elasticsearch-haystack

사용법 (Usage)

단독 사용 (On its own)

인덱스에 몇 개의 문서를 쓴 뒤 SQL 쿼리를 실행해 봐요. 아래 예시는 인덱스에서 content 필드를 선택하고 반환된 열과 행을 읽어요.

from haystack import Document
from haystack_integrations.components.retrievers.elasticsearch import (
    ElasticsearchSQLRetriever,
)
from haystack_integrations.document_stores.elasticsearch import (
    ElasticsearchDocumentStore,
)
from haystack.document_stores.types import DuplicatePolicy

document_store = ElasticsearchDocumentStore(
    hosts="http://localhost:9200/", index="my_index"
)

documents = [
    Document(content="There are over 7,000 languages spoken around the world today."),
    Document(
        content="Elephants have been observed to behave in a way that indicates a high level of self-awareness, such as recognizing themselves in mirrors.",
    ),
    Document(
        content="In certain parts of the world, like the Maldives, Puerto Rico, and San Diego, you can witness the phenomenon of bioluminescent waves.",
    ),
]

# DuplicatePolicy.SKIP is optional, but useful to run the script multiple times without throwing errors
document_store.write_documents(documents=documents, policy=DuplicatePolicy.SKIP)

retriever = ElasticsearchSQLRetriever(document_store=document_store)
output = retriever.run(query='SELECT content FROM "my_index" LIMIT 10')

result = output["result"]
print(result["columns"])  # column metadata, e.g. [{"name": "content", "type": "text"}]
for row in result["rows"]:
    print(row)

집계 쿼리 실행 (Running an aggregation query)

이 컴포넌트는 원시 SQL 응답을 반환하므로, 문서 기반 리트리버가 지원하지 않는 집계(예: 문서 개수 세기)에도 사용할 수 있어요.

retriever = ElasticsearchSQLRetriever(document_store=document_store)
output = retriever.run(query='SELECT COUNT(*) AS doc_count FROM "my_index"')

result = output["result"]
print(result["rows"])  # e.g. [[3]]

잘못된 형식이거나 실패하는 쿼리에서 예외가 발생하지 않게 하려면 raise_on_failure=False로 컴포넌트를 초기화하세요. 이 경우 실패한 쿼리는 경고를 기록하고 빈 딕셔너리를 반환해요.

ElasticsearchSQLRetriever