Azure Cosmos DB의 트랜잭션 일괄 처리 작업

아티클
08/15/2024

적용 대상: NoSQL

트랜잭션 일괄 처리는 성공 또는 실패해야 하는 지점 작업 그룹과 컨테이너의 동일한 파티션 키를 함께 제공합니다. 작업이 정의되고 일괄 처리에 추가되며 일괄 처리가 실행됩니다. 트랜잭션 일괄 처리 작업 내에 설명된 순서대로 모든 작업이 성공적으로 수행되면 트랜잭션이 커밋됩니다. 그러나 어느 한 작업이라도 실패하면 전체 트랜잭션이 롤백됩니다.

Azure Cosmos DB에서 트랜잭션이란?

일반적인 데이터베이스의 트랜잭션은 하나의 논리적 작업 단위로 수행되는 작업 시퀀스로 정의할 수 있습니다. 각 트랜잭션은 ACID(원자성, 일관성, 격리, 내구성) 속성을 보장합니다.

원자성은 트랜잭션 내부에서 수행된 모든 작업이 하나의 단위로 처리되어 모두 커밋되거나 커밋되지 않도록 합니다.
일관성은 데이터가 트랜잭션 간에 항상 올바른 상태로 유지되도록 합니다.
격리는 두 트랜잭션이 서로를 방해하지 않도록 합니다. 많은 상용 시스템은 애플리케이션 요구에 따라 사용할 수 있는 여러 격리 수준을 제공합니다.
내구성은 데이터베이스에서 커밋된 변경 내용이 항상 유지되도록 합니다. Azure Cosmos DB는 동일한 논리 파티션 키 내의 작업에 대해 스냅샷 격리를 사용하는 완전한 ACID 규격 트랜잭션을 지원합니다.

트랜잭션 일괄 처리 작업 및 저장 프로시저

현재, Azure Cosmos DB는 작업에 대한 트랜잭션 범위도 제공하는 저장 프로시저를 지원합니다. 그러나 트랜잭션 일괄 처리 작업은 다음과 같은 이점을 제공합니다.

언어 옵션 – 트랜잭션 일괄 처리는 이미 작업 중인 SDK 및 언어에서 지원되지만 저장 프로시저는 JavaScript로 작성해야 합니다.
코드 버전 관리 - 애플리케이션 코드의 버전을 관리하고 CI/CD 파이프라인에 온보딩하는 것은 저장 프로시저의 업데이트를 오케스트레이션하고 적절한 시간에 롤오버를 수행하는 것보다 훨씬 자연스럽게 진행됩니다. 또한 변경 내용을 보다 쉽게 롤백할 수 있습니다.
성능 - 저장 프로시저 실행과 비교할 때 해당 작업에 대한 대기 시간이 최대 30% 줄어듭니다.
콘텐츠 serialization – 트랜잭션 일괄 처리 내의 각 작업은 해당 페이로드에 대한 사용자 지정 serialization 옵션을 사용할 수 있습니다.

트랜잭션 일괄 처리 작업을 만드는 방법

트랜잭션 일괄 처리 작업을 만들 때 컨테이너 인스턴스로 시작하고 CreateTransactionalBatch를 호출합니다.

PartitionKey partitionKey = new PartitionKey("road-bikes");

TransactionalBatch batch = container.CreateTransactionalBatch(partitionKey);

다음으로 일괄 처리에 여러 작업을 추가합니다.

Product bike = new (
    id: "68719520766",
    category: "road-bikes",
    name: "Chropen Road Bike"
);

batch.CreateItem<Product>(bike);

Part part = new (
    id: "68719519885",
    category: "road-bikes",
    name: "Tronosuros Tire",
    productId: bike.id
);

batch.CreateItem<Part>(part);

마지막으로 일괄 처리에서 ExecuteAsync를 호출합니다.

using TransactionalBatchResponse response = await batch.ExecuteAsync();

응답이 수신되면 응답이 성공적으로 수행되는지 검토합니다. 응답이 성공을 나타내는 경우 결과를 추출합니다.

if (response.IsSuccessStatusCode)
{
    TransactionalBatchOperationResult<Product> productResponse;
    productResponse = response.GetOperationResultAtIndex<Product>(0);
    Product productResult = productResponse.Resource;

    TransactionalBatchOperationResult<Part> partResponse;
    partResponse = response.GetOperationResultAtIndex<Part>(1);
    Part partResult = partResponse.Resource;
}

Important

오류가 발생하는 경우 실패한 작업에 해당 오류에 대한 상태 코드가 포함됩니다. 다른 모든 작업에는 424 상태 코드(실패한 종속성)가 포함됩니다. 이미 존재하는 항목을 만들려고 하여 작업이 실패하면 상태 코드 409(충돌)가 반환됩니다. 이 상태 코드를 사용하여 트랜잭션 오류의 원인을 식별할 수 있습니다.

트랜잭션 일괄 처리 작업을 만들 때 CosmosBatch.createCosmosBatch를 호출합니다.

PartitionKey partitionKey = new PartitionKey("road-bikes");

CosmosBatch batch = CosmosBatch.createCosmosBatch(partitionKey);

다음으로 일괄 처리에 여러 작업을 추가합니다.

Product bike = new Product();
bike.setId("68719520766");
bike.setCategory("road-bikes");
bike.setName("Chropen Road Bike");

batch.createItemOperation(bike);

Part part = new Part();
part.setId("68719519885");
part.setCategory("road-bikes");
part.setName("Tronosuros Tire");
part.setProductId(bike.getId());

batch.createItemOperation(part);

마지막으로 컨테이너 인스턴스를 사용하여 일괄 처리로 executeCosmosBatch를 호출합니다.

CosmosBatchResponse response = container.executeCosmosBatch(batch);

응답이 수신되면 응답이 성공적으로 수행되는지 검토합니다. 응답이 성공을 나타내는 경우 결과를 추출합니다.

if (response.isSuccessStatusCode())
{
    List<CosmosBatchOperationResult> results = response.getResults();
}

Important

컨테이너 인스턴스를 가져오거나 만듭니다.

container = database.create_container_if_not_exists(id="batch_container",
                                                        partition_key=PartitionKey(path='/category'))

Python에서 트랜잭션 Batch 작업은 단일 작업 API와 매우 유사하며, (Operation_type_string, args_tuple, Batch_Operation_kwargs_dictionary)를 포함하는 튜플입니다. 다음은 일괄 처리 작업 기능을 시연하는 데 사용되는 샘플 항목입니다.


create_demo_item = {
    "id": "68719520766",
    "category": "road-bikes",
    "name": "Chropen Road Bike"
}

# for demo, assume that this item already exists in the container. 
# the item id will be used for read operation in the batch
read_demo_item1 = {
    "id": "68719519884",
    "category": "road-bikes",
    "name": "Tronosuros Tire",
    "productId": "68719520766"
}

# for demo, assume that this item already exists in the container. 
# the item id will be used for read operation in the batch
read_demo_item2 = {
    "id": "68719519886",
    "category": "road-bikes",
    "name": "Tronosuros Tire",
    "productId": "68719520766"
}

# for demo, assume that this item already exists in the container. 
# the item id will be used for read operation in the batch
read_demo_item3 = {
    "id": "68719519887",
    "category": "road-bikes",
    "name": "Tronosuros Tire",
    "productId": "68719520766"
}

# for demo, we'll upsert the item with id 68719519885
upsert_demo_item = {
    "id": "68719519885",
    "category": "road-bikes",
    "name": "Tronosuros Tire Upserted",
    "productId": "68719520768"
}

# for replace demo, we'll replace the read_demo_item2 with this item
replace_demo_item = {
    "id": "68719519886",
    "category": "road-bikes",
    "name": "Tronosuros Tire replaced",
    "productId": "68719520769"
}

# for replace with etag match demo, we'll replace the read_demo_item3 with this item
# The use of etags and if-match/if-none-match options allows users to run conditional replace operations
# based on the etag value passed. When using if-match, the request will only succeed if the item's latest etag
# matches the passed in value. For more on optimistic concurrency control, see the link below:
# https://learn.microsoft.com/azure/cosmos-db/nosql/database-transactions-optimistic-concurrency
replace_demo_item_if_match_operation = {
    "id": "68719519887",
    "category": "road-bikes",
    "name": "Tronosuros Tireh",
    "wasReplaced": "Replaced based on etag match"
    "productId": "68719520769"
}

일괄 처리에 추가할 작업을 준비합니다.

create_item_operation = ("create", (create_demo_item,), {})
read_item_operation = ("read", ("68719519884",), {})
delete_item_operation = ("delete", ("68719519885",), {})
upsert_item_operation = ("upsert", (upsert_demo_item,), {})
replace_item_operation = ("replace", ("68719519886", replace_demo_item), {})
replace_item_if_match_operation = ("replace",
                                       ("68719519887", replace_demo_item_if_match_operation),
                                       {"if_match_etag": container.client_connection.last_response_headers.get("etag")})

일괄 처리에 작업을 추가합니다.

batch_operations = [
        create_item_operation,
        read_item_operation,
        delete_item_operation,
        upsert_item_operation,
        replace_item_operation,
        replace_item_if_match_operation
    ]

마지막으로 일괄 처리를 실행합니다.

try:
    # Run that list of operations
    batch_results = container.execute_item_batch(batch_operations=batch_operations, partition_key="road_bikes")
    # Batch results are returned as a list of item operation results - or raise a CosmosBatchOperationError if
    # one of the operations failed within your batch request.
    print("\nResults for the batch operations: {}\n".format(batch_results))
except exceptions.CosmosBatchOperationError as e:
        error_operation_index = e.error_index
        error_operation_response = e.operation_responses[error_operation_index]
        error_operation = batch_operations[error_operation_index]
        print("\nError operation: {}, error operation response: {}\n".format(error_operation, error_operation_response))
    # [END handle_batch_error]

일괄 처리에서 패치 작업 및 바꾸기_if_match_etag 작업 사용에 대한 참고 사항
일괄 처리 작업 kwargs 사전은 제한되어 있으며 총 3개의 서로 다른 키 값만 사용합니다. 일괄 처리 내에서 조건부 패치를 사용하려는 경우 패치 작업에는 filter_predicate 키를 사용할 수 있으며, 작업에 etag를 사용하려는 경우 if_match_etag/if_none_match_etag 키도 사용할 수 있습니다.
batch_operations = [
       ("replace", (item_id, item_body), {"if_match_etag": etag}),
       ("patch", (item_id, operations), {"filter_predicate": filter_predicate, "if_none_match_etag": etag}),
   ]

Important

오류가 발생하는 경우 실패한 작업에 해당 오류에 대한 상태 코드가 포함됩니다. 다른 모든 작업에는 424 상태 코드(실패한 종속성)가 포함됩니다. 이미 존재하는 항목을 만들려고 하여 작업이 실패하면 상태 코드 409(충돌)가 반환됩니다. 이 상태 코드를 사용하여 트랜잭션 오류의 원인을 식별할 수 있습니다.

트랜잭션 일괄 처리 작업을 실행하는 방법

트랜잭션 Batch가 실행되면 트랜잭션 Batch의 모든 작업이 그룹화되고 단일 페이로드로 직렬화되며 단일 요청으로 Azure Cosmos DB 서비스에 전송됩니다.

이 서비스는 요청을 수신하고 트랜잭션 범위 내에서 모든 작업을 실행하며 동일한 serialization 프로토콜을 사용하여 응답을 반환합니다. 이 응답은 성공 또는 실패이며 작업당 개별 작업 응답을 제공합니다.

SDK는 결과를 확인하고 필요에 따라 각 내부 작업 결과를 추출하기 위한 응답을 제공합니다.

제한 사항

현재는 다음과 같은 두 가지 알려진 제한 사항이 있습니다.

Azure Cosmos DB 요청 크기 제한은 트랜잭션 Batch 페이로드의 크기를 2MB를 초과하지 않도록 제한하고 최대 실행 시간은 5초입니다.
성능이 예상대로 SLA 내에 있는지 확인하기 위해 현재 트랜잭션 Batch당 작업 100개로 제한되어 있습니다.

다음 단계

TransactionalBatch로 수행할 수 있는 작업에 대해 자세히 알아봅니다.
Azure Cosmos DB .NET SDK를 사용하는 방법에 대한 자세한 내용은 샘플을 참조하세요.

다음을 통해 공유

Azure Cosmos DB의 트랜잭션 일괄 처리 작업

Azure Cosmos DB에서 트랜잭션이란?

트랜잭션 일괄 처리 작업 및 저장 프로시저

트랜잭션 일괄 처리 작업을 만드는 방법

트랜잭션 일괄 처리 작업을 실행하는 방법

제한 사항

다음 단계

피드백

추가 리소스