dm-limits-and-best-practices은(는) 무엇을 하나요?
CDF Data Modeling API 모범 사례를 위한 참조 스킬입니다. 동시성 제한(429 오류 방지), instances.list 및 instances.query에 대한 페이지네이션 패턴, 쓰기 작업 배치, 검색 vs 필터 가이드, 그리고 동시 요청을 제어하기 위한 QueuedTaskRunner(세마포어) 유틸리티를 다룹니다.
트리거: DMS 제한, 429 오류, 속도 제한(rate limit), 페이지네이션(pagination), 커서(cursor), nextCursor, 배치(batching), 세마포어(semaphore), QueuedTaskRunner, cdfTaskRunner, instances.search, instances.list, instances.query, instances.upsert, 동시성(concurrency), 데드락(deadlock).
동시성 제한 및 429 오류 방지
CDF API에는 호출당 허용되는 최대 동시 요청 수가 있습니다. 이 제한을 초과하면 429 Too Many Requests 응답이 반환됩니다.
- 기본 동시성 제한: API 엔드포인트당 10개의 동시 요청입니다.
- 429 응답 처리: 429 오류가 발생하면 지수 백오프(Exponential Backoff)와 재시도 로직을 구현해야 합니다.
- 권장사항:
QueuedTaskRunner(아래 참조)를 사용하여 동시 요청을 제한함으로써 429 오류를 사전에 방지하세요.
페이지네이션 패턴
instances.list
instances.list는 결과를 페이지 단위로 반환합니다. 각 응답에는 다음 페이지를 가져오는 데 사용할 수 있는 nextCursor 필드가 포함됩니다.
# 예시: instances.list 페이지네이션
cursor = None
while True:
response = client.data_modeling.instances.list(
space="my_space",
limit=1000,
cursor=cursor
)
# 결과 처리
for instance in response.instances:
process(instance)
cursor = response.nextCursor
if cursor is None:
break
instances.query
instances.query도 페이지네이션을 지원합니다. 쿼리 결과가 큰 경우 nextCursor를 사용하여 모든 결과를 반복하세요.
# 예시: instances.query 페이지네이션
query = {
"with": {
"items": {
"nodes": {
"filter": {
"equals": {
"property": ["node", "type"],
"value": {"space": "my_space", "externalId": "MyType"}
}
},
"limit": 100
}
}
},
"select": {
"items": []
}
}
cursor = None
while True:
if cursor:
query["cursor"] = cursor
response = client.data_modeling.instances.query(query)
# 결과 처리
for item in response.items:
process(item)
cursor = response.nextCursor
if cursor is None:
break
쓰기 작업 배치
대량의 인스턴스를 생성하거나 업데이트할 때는 단일 요청으로 처리하는 것보다 배치로 나누는 것이 좋습니다.
- 배치 크기 권장: 요청당 최대 1000개의 인스턴스
- 동시 배치:
QueuedTaskRunner를 사용하여 여러 배치를 동시에 전송하되, 전체 동시성 제한을 초과하지 않도록 하세요.
from cognite.client.data_classes import InstanceWrite
def batch_upsert_instances(client, instances: list[InstanceWrite], batch_size=1000, max_concurrency=5):
"""인스턴스를 배치로 upsert하고 동시성을 제어합니다."""
from cognite.client.utils import QueuedTaskRunner
runner = QueuedTaskRunner(max_concurrency=max_concurrency)
for i in range(0, len(instances), batch_size):
batch = instances[i:i + batch_size]
runner.submit(client.data_modeling.instances.upsert, batch)
runner.wait_for_completion()
검색 vs 필터 가이드
instances.search: 텍스트 검색 및 퍼지 매칭(fuzzy matching)에 사용합니다. 특정 속성 값에 대한 전체 텍스트 검색이 필요할 때 적합합니다.instances.filter: 속성 값에 대한 정확한 일치, 범위 쿼리, 존재 여부 확인 등 구조화된 필터링에 사용합니다. 검색보다 성능이 뛰어나며 대규모 데이터셋에 권장됩니다.
일반 규칙: 가능하면 항상 filter를 사용하고, 텍스트 기반 검색이 정말 필요할 때만 search를 사용하세요.
QueuedTaskRunner (세마포어) 유틸리티
QueuedTaskRunner는 동시 실행되는 태스크 수를 제한하는 유틸리티입니다. cdfTaskRunner로도 알려져 있습니다.
주요 특징
- 동시 요청 수를 사용자 정의 가능한 한도로 제한합니다.
- 완료될 때까지 태스크를 자동으로 큐에 대기시킵니다.
- 예외를 포착하고 보고합니다.
사용 예시
from cognite.client.utils import QueuedTaskRunner
# 최대 5개의 동시 요청으로 러너 생성
runner = QueuedTaskRunner(max_concurrency=5)
# 태스크 제출 (즉시 실행되지 않고 큐에 추가됨)
for item in items:
runner.submit(process_item, item)
# 모든 태스크가 완료될 때까지 대기
runner.wait_for_completion()
# 예외 확인
if runner.exceptions:
for exc in runner.exceptions:
print(f"태스크 실패: {exc}")
전체 예시: 안전한 대량 Upsert
다음은 모든 모범 사례를 결합한 종합 예시입니다.
from cognite.client import CogniteClient
from cognite.client.data_classes import InstanceWrite
from cognite.client.utils import QueuedTaskRunner
client = CogniteClient()
# 업서트할 인스턴스 목록 (예시)
instances = [...] # 많은 인스턴스
BATCH_SIZE = 500
MAX_CONCURRENCY = 5
def upsert_batch(batch):
"""단일 배치를 upsert하고 결과를 반환합니다."""
return client.data_modeling.instances.upsert(batch)
runner = QueuedTaskRunner(max_concurrency=MAX_CONCURRENCY)
# 인스턴스를 배치로 나누고 태스크 제출
for i in range(0, len(instances), BATCH_SIZE):
batch = instances[i:i + BATCH_SIZE]
runner.submit(upsert_batch, batch)
# 완료 대기
runner.wait_for_completion()
# 오류 처리
if runner.exceptions:
print(f"다음 개수의 배치가 실패했습니다: {len(runner.exceptions)}")
for exc in runner.exceptions:
print(f" 오류: {exc}")
else:
print(f"모든 {len(instances)}개 인스턴스가 성공적으로 upsert되었습니다.")