CSVDocumentSplitter

CSVDocumentSplitter

CSVDocumentSplitter는 CSV 형식의 내용을 담은 문서 리스트를 받아, 각각이 원본 문서에서 추출된 하위 테이블(sub-table)을 나타내는 새 Document 객체 리스트를 반환해요.

출처: 문서

본문

Overview

CSVDocumentSplitter는 CSV 형식의 내용을 담은 문서 리스트를 받아, 각각이 원본 문서에서 추출된 하위 테이블을 나타내는 새 Document 객체 리스트를 반환합니다.

스플리터에는 두 가지 작동 모드가 있어요:

  • threshold(기본값): 주어진 임계값을 초과하는 빈 행이나 열을 식별해 문서를 그에 따라 분할합니다.
  • row-wise: 각 행을 독립된 하위 테이블로 취급해 별도 문서로 분할합니다.

분할 과정은 다음 규칙을 따릅니다:

  • 행 기반 분할: row_split_threshold가 설정되면, 이 임계값과 같거나 초과하는 연속적인 빈 행이 분할을 일으켜요.
  • 열 기반 분할: column_split_threshold가 설정되면, 이 임계값과 같거나 초과하는 연속적인 빈 열이 분할을 일으켜요.
  • 재귀 분할: 두 임계값이 모두 제공되면, CSVDocumentSplitter는 먼저 행 기준으로 분할하고 그다음 열 기준으로 분할합니다. 더 많은 빈 행이 감지되면 분할 과정이 다시 호출됩니다. 이는 하위 테이블이 완전히 분리되도록 보장합니다.

추출된 각 하위 테이블은 원본 문서의 메타데이터를 유지하고 추가 필드를 포함합니다:

  • source_id — 원본 문서의 ID
  • row_idx_start — 원본 문서에서 하위 테이블의 시작 행 인덱스
  • col_idx_start — 원본 문서에서 하위 테이블의 시작 열 인덱스
  • split_id — 문서 내 분할의 순차 ID

이 컴포넌트는 구조화된 데이터를 효율적으로 추출·저장해야 하는 문서 처리 파이프라인에서 특히 유용해요.

Supported Document Stores

CSVDocumentSplitter는 다음 Document Store와 호환됩니다:

AstraDocumentStore, ChromaDocumentStore, ElasticsearchDocumentStore, OpenSearchDocumentStore, PgvectorDocumentStore, PineconeDocumentStore, QdrantDocumentStore, WeaviateDocumentStore, MilvusDocumentStore, Neo4jDocumentStore

Usage

On its own

파이프라인 밖에서도 CSVDocumentSplitter를 써서 CSV 문서를 직접 처리할 수 있어요:

from haystack import Document
from haystack.components.preprocessors import CSVDocumentSplitter

splitter = CSVDocumentSplitter(row_split_threshold=1, column_split_threshold=1)

doc = Document(
    content="""ID,LeftVal,,,RightVal,Extra
1,Hello,,,World,Joined
2,StillLeft,,,StillRight,Bridge
,,,,,
A,B,,,C,D
E,F,,,G,H
""",
)
split_result = splitter.run([doc])
print(split_result["documents"])  # List of split tables as Documents

In a pipeline

CSVDocumentSplitter를 Haystack 인덱싱 파이프라인에 통합하는 예시예요:

from haystack import Pipeline, Document
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.converters.csv import CSVToDocument
from haystack.components.preprocessors import CSVDocumentSplitter
from haystack.components.preprocessors import CSVDocumentCleaner
from haystack.components.writers import DocumentWriter

# Initialize components
document_store = InMemoryDocumentStore()
p = Pipeline()
p.add_component(instance=CSVToDocument(), name="csv_file_converter")
p.add_component(instance=CSVDocumentSplitter(), name="splitter")
p.add_component(instance=CSVDocumentCleaner(), name="cleaner")
p.add_component(instance=DocumentWriter(document_store=document_store), name="writer")

# Connect components
p.connect("csv_file_converter.documents", "splitter.documents")
p.connect("splitter.documents", "cleaner.documents")
p.connect("cleaner.documents", "writer.documents")

# Run pipeline
p.run({"csv_file_converter": {"sources": ["path/to/your/file.csv"]}})

이 파이프라인은 CSV 내용을 추출하고, 구조화된 하위 테이블로 분할하고, 빈 행·열을 제거해 CSV 문서를 정리한 뒤, 결과 문서를 추후 검색·처리를 위해 Document Store에 저장해요.

더 알아보기 (Learn more)