배치 임포트
배치 임포트 (Batch import)
데이터를 하나씩 넣으면 대규모 데이터를 다룰 때 너무 느려요. Weaviate의 **배치 임포트(Batch import)**는 여러 객체와 교차 참조를 한 번에 효율적으로 추가하는 방법입니다. 대부분의 용도에는 **서버 사이드 배칭(server-side batching)**을 권장합니다 — 서버가 클라이언트에게 다음에 얼마나 보낼지 알려주기 때문에 배치 파라미터를 직접 조율하지 않아도 되거든요.
서버 사이드 배칭(Server-side batching)
서버 사이드(또는 "자동") 배칭에서는 클라이언트가 서버의 피드백에 따라 정해진 배치 크기로 데이터를 보냅니다. 코드가 단순해지고 서버가 자기 부하를 스스로 관리할 수 있어요. 진입점은 두 가지입니다.
- 데이터 소스에서 스트리밍(대규모 데이터셋 권장) — 소스에서 읽으면서 객체를 하나씩 추가해 전체 데이터셋이 메모리에 안 들어가도 되게 합니다.
- 메모리 내 리스트 인제스트 — 이미 메모리에 들고 있는 객체 목록을 단일 호출로 임포트합니다.
서버 사이드 배칭은 gRPC API를 사용하며, 현재 클라이언트 버전에서는 기본 활성화입니다.
MyCollection 컬렉션에 객체를 추가하는 예시입니다.
스트리밍(stream)
Python에서는 batch.stream() 컨텍스트 매니저를 열고 객체를 하나씩 추가합니다. 클라이언트는 서버가 요청하는 속도로 보내요. async Python 클라이언트도 stream()과 일회성 ingest()로 서버 사이드 배칭을 지원합니다.
data_rows = [
{"title": f"Object {i+1}"} for i in range(5)
]
collection = client.collections.use("MyCollection")
# 서버 사이드 배칭에는 `stream`을 씁니다.
# 클라이언트는 서버가 지정한 속도로 배치 단위로 데이터를 보냅니다.
with collection.batch.stream() as batch:
for data_row in data_rows:
batch.add_object(
properties=data_row,
)
if batch.number_errors > 10:
print("Batch import stopped due to excessive errors.")
break
failed_objects = collection.batch.failed_objects
if failed_objects:
print(f"Number of failed imports: {len(failed_objects)}")
print(f"First failed object: {failed_objects[0]}")
소스 파일을 레코드 단위로 읽는 제너레이터를 넘기면 data.ingest()로 데이터 소스에서 스트리밍할 수도 있어요. 객체는 제너레이터가 생성하는 대로 서버로 가므로 소스가 메모리에 안 들어가도 됩니다.
import json
# 소스 파일의 각 줄은 JSON 객체 하나를 담고 있습니다
def read_objects(path):
with open(path) as f:
for line in f:
line = line.strip()
if not line: # 빈 줄 건너뛰기
continue
record = json.loads(line)
yield {"title": record["title"]}
collection = client.collections.use("MyCollection")
# `ingest`는 진행하면서 제너레이터에서 객체를 가져옵니다
result = collection.data.ingest(read_objects("my-data.jsonl"))
if result.errors:
print(f"Number of failed imports: {len(result.errors)}")
TypeScript에서는 data.ingest()가 서버 사이드 배칭 API이며 별도의 스트리밍 컨텍스트가 없습니다. 어떤 Iterable이든 받아서, 제너레이터를 넘기면 전체 리스트를 메모리에 쌓지 않고 객체를 서버로 스트리밍합니다.
참고로 Go 클라이언트는 서버 사이드 배칭을 지원하지 않으므로 매뉴얼 배칭을 써야 합니다.
인메모리 리스트 인제스트
객체가 이미 메모리에 있다면 전체 리스트를 단일 호출로 임포트할 수 있습니다. 클라이언트가 서버 사이드 배칭으로 보내므로, 한 요청으로 보내면 서버의 GRPC_MAX_MESSAGE_SIZE 한도를 초과할 큰 리스트에서도 안전합니다.
data_rows = [
{"title": f"Object {i+1}"} for i in range(5)
]
collection = client.collections.use("MyCollection")
# `ingest`는 서버 사이드 배칭으로 전체 리스트를 단일 호출로 임포트합니다
result = collection.data.ingest(data_rows)
# 반환 객체는 `insert_many`와 같습니다
if result.errors:
print(f"Number of failed imports: {len(result.errors)}")
# `errors`는 실패한 객체의 인덱스를 키로 하는 dict입니다
for index, error in result.errors.items():
print(f"Failed object at index {index}: {error.message}")
Python에서 data.ingest()는 큰 리스트를 단일 요청으로 보내는 insert_many의 안전한 대체재입니다. ingest는 일반 속성 dict 또는 DataObject 인스턴스(객체 ID·벡터·참조 설정용)를 받고 insert_many와 같은 반환 객체를 줍니다.
매뉴얼 배칭(Manual batching)
배치 크기와 동시성을 직접 제어하고 싶거나, 서버 사이드 배칭을 아직 지원하지 않는 클라이언트(예: Go)를 쓴다면 매뉴얼(클라이언트 사이드) 배칭을 사용합니다.
data_rows = [
{"title": f"Object {i+1}"} for i in range(5)
]
collection = client.collections.use("MyCollection")
with collection.batch.fixed_size(batch_size=200) as batch:
for data_row in data_rows:
batch.add_object(
properties=data_row,
)
if batch.number_errors > 10:
print("Batch import stopped due to excessive errors.")
break
failed_objects = collection.batch.failed_objects
if failed_objects:
print(f"Number of failed imports: {len(failed_objects)}")
print(f"First failed object: {failed_objects[0]}")
오류 처리
배치 임포트는 객체별로 실패를 보고합니다. 한 객체의 문제가 나머지 임포트를 중단하지 않아요. 오류는 서버 사이드와 매뉴얼 배칭에서 동일하게 보고됩니다. import 도중과 이후에 실패 항목을 검사해 데이터 문제를 일찍 잡으세요.
Python의 경우:
- 배칭 컨텍스트 매니저 안에서
batch.number_errors가 실패한 객체·참조의 누적 개수를 담습니다. 이 카운터로 임포트를 중단하고 원인을 조사할 수 있어요. - 컨텍스트가 닫힌 뒤에는
collection.batch.failed_objects,collection.batch.failed_references가 실패 항목을 담습니다. - 일회성
data.ingest()는insert_many와 같은 결과 객체를 반환합니다.errorsdict가 각 실패 객체의 원래 인덱스를 오류에 매핑합니다.
자세한 오류 처리는 Python 클라이언트 레퍼런스 페이지에서 확인하세요.
임포트 객체 커스터마이즈
배치로 임포트한 객체는 개별 생성 객체와 같은 파라미터를 지원합니다. 커스텀 ID, 벡터, 교차 참조가 그 예입니다.
ID 값 지정
Weaviate는 각 객체에 UUID를 생성합니다. 객체 ID는 고유해야 하므로, ID를 직접 설정한다면 중복을 막기 위해 결정적 UUID 메서드를 씁니다 — Python의 generate_uuid5, TypeScript의 generateUuid5.
from weaviate.util import generate_uuid5 # 결정적 ID 생성
from weaviate.classes.data import DataObject
data_rows = [{"title": f"Object {i+1}"} for i in range(5)]
collection = client.collections.use("MyCollection")
data_objects = [
DataObject(
properties=data_row,
uuid=generate_uuid5(data_row)
)
for data_row in data_rows
]
result = collection.data.ingest(data_objects)
if result.errors:
print(f"Number of failed imports: {len(result.errors)}")
벡터 지정
vector 속성으로 각 객체의 벡터를 지정합니다.
from weaviate.classes.data import DataObject
data_rows = [{"title": f"Object {i+1}"} for i in range(5)]
vectors = [[0.1] * 1536 for i in range(5)]
collection = client.collections.use("MyCollection")
data_objects = [
DataObject(
properties=data_row,
vector=vectors[i]
)
for i, data_row in enumerate(data_rows)
]
result = collection.data.ingest(data_objects)
if result.errors:
print(f"Number of failed imports: {len(result.errors)}")
네임드 벡터 지정
컬렉션에 네임드 벡터가 구성되어 있다면 객체 생성 시 네임드 벡터를 지정할 수 있습니다.
from weaviate.classes.data import DataObject
data_rows = [{
"title": f"Object {i+1}",
"body": f"Body {i+1}"
} for i in range(5)]
title_vectors = [[0.12] * 1536 for _ in range(5)]
body_vectors = [[0.34] * 1536 for _ in range(5)]
collection = client.collections.use("MyCollection")
data_objects = [
DataObject(
properties=data_row,
vector={
"title": title_vectors[i],
"body": body_vectors[i],
}
)
for i, data_row in enumerate(data_rows)
]
result = collection.data.ingest(data_objects)
if result.errors:
print(f"Number of failed imports: {len(result.errors)}")
참조(reference)와 함께 임포트
교차 참조를 통해 한 객체에서 다른 객체로의 링크를 배치 생성할 수 있습니다.
from weaviate.classes.data import DataObject
collection = client.collections.use("Author")
data_objects = [
DataObject(
properties={"name": "Jane Austen"},
references={"writesFor": target_uuid},
),
]
result = collection.data.ingest(data_objects)
if result.errors:
print(f"Number of failed imports: {len(result.errors)}")
큰 파일에서 데이터 스트리밍하기
데이터셋이 메모리에 안 들어간다면 한 번에 전부 로드하지 말고, 소스 파일을 지연(lazy) 읽기하며 import에 객체를 추가하세요.
- 서버 사이드 스트리밍 컨텍스트에서는 파일에서 읽으면서 객체를 추가합니다. 클라이언트는 서버가 요청하는 속도로 보내므로 메모리 사용량이 일정합니다.
- Python·TypeScript의 일회성 임포트는 어떤 iterable이든 받으므로, 파일을 레코드 단위로 읽는 제너레이터 같은 지연 소스를 넘길 수 있습니다.
- 매뉴얼 배칭에서도 같은 패턴을 적용합니다 — 읽으면서 배치에 추가하세요.
JSON 파일은 한 번에 하나씩 객체를 생성하는 스트리밍 파서(예: Python의 ijson)를 쓰고, CSV 파일은 chunksize 파라미터를 쓰는 pandas처럼 청크 단위로 읽는 게 좋습니다.
추가 고려 사항
데이터 임포트는 리소스가 많이 필요할 수 있어요. 많은 양의 데이터를 임포트할 때는 다음을 고려하세요.
- 비동기 임포트 — import 속도를 최대화하려면 Weaviate 구성에서
ASYNC_INDEXING환경 변수를true로 설정해 비동기 인덱싱을 켜세요. 이렇게 하면 벡터 인덱스 구축을 객체 생성에서 분리해 인덱스 구축 때문에 import가 느려지지 않습니다. - 새 테넌트 자동 추가 — 자동 테넌트 생성에 대한 자세한 내용은 멀티테넌시의 자동 테넌트 생성을 참고하세요.
큰 import에서 배치 방식이 서버 메모리를 압박한다면 fixed_size 대신 서버 사이드 stream() 또는 ingest()를 쓰면 서버가 부하를 관리하게 할 수 있어요.