Operações de lote transacionais no Azure Cosmos DB

APLICA-SE A: NoSQL

O lote transacional descreve um grupo de operações pontuais que precisam ou ter sucesso ou falhar juntas com a mesma chave de partição num recipiente. As operações são definidas, adicionadas ao lote, e o lote é executado. Se todas as operações forem bem-sucedidas na ordem em que estão descritas na operação em lote transacional, a transação será confirmada. No entanto, se alguma operação falhar, toda a transação é revertida.

O que é uma transação no Azure Cosmos DB

Uma transação numa base de dados típica pode ser definida como uma sequência de operações executadas como uma única unidade lógica de trabalho. Cada transação garante as propriedades ACID (Atomicidade, Consistência, Isolamento, Durabilidade).

• Atomicidade garante que todas as operações realizadas dentro de uma transação sejam tratadas como uma única unidade, e que ou todas sejam confirmadas ou nenhuma seja.
• Consistência garante que os dados estejam sempre em um estado válido em todas as transações.
• Isolamento garante que nenhuma transação interfira com outra – muitos sistemas comerciais oferecem vários níveis de isolamento que podem ser usados conforme as necessidades da aplicação.
• Durabilidade garante que qualquer alteração que seja confirmada numa base de dados estará sempre presente. Azure Cosmos DB suporta transações totalmente compatíveis com ACID com isolamento de instantâneo para operações dentro da mesma chave de partição lógica.

Operações em lote transacionais e procedimentos armazenados

Atualmente, o Azure Cosmos DB dá suporte a procedimentos armazenados, que também fornecem o escopo transacional em operações. No entanto, operações em lote transacionais oferecem os seguintes benefícios:

• Opção de idioma – O lote transacional é suportado no SDK e na linguagem com que você já trabalha, enquanto os procedimentos armazenados precisam ser escritos em JavaScript.
• Controle de versão de código – A versão do código da aplicação e a sua entrada no pipeline de CI/CD é muito mais natural do que orquestrar a atualização de um procedimento armazenado e garantir que a transição ocorra no momento certo. Também facilita a reversão de alterações.
• Desempenho – Reduziu a latência em operações equivalentes em até 30% em comparação com a execução do procedimento armazenado.
• Serialização de conteúdo – Cada operação dentro de um lote transacional pode usar opções de serialização personalizadas para a sua carga útil.

Como criar uma operação em lote transacional

.NET

Ao criar uma operação em lote transacional, comece com uma instância de contêiner e chame CreateTransactionalBatch:

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

TransactionalBatch batch = container.CreateTransactionalBatch(partitionKey);

Em seguida, adicione várias operações ao lote:

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);

Finalmente, chame ExecuteAsync no lote:

using TransactionalBatchResponse response = await batch.ExecuteAsync();

Assim que a resposta for recebida, verifique se a resposta é bem-sucedida. Se a resposta indicar um sucesso, extraia os resultados.

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;
}

Importante

Se houver uma falha, a operação com falha terá um código de status de seu erro correspondente. Todas as outras operações terão um código de status 424 (dependência falhada). Se a operação falhar porque tenta criar um item que já existe, será retornado um código de estado 409 (conflito). O código de estado permite identificar a causa da falha da transação.

Java

Ao criar uma operação de lote transacional, chame CosmosBatch.createCosmosBatch:

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

CosmosBatch batch = CosmosBatch.createCosmosBatch(partitionKey);

Em seguida, adicione várias operações ao lote:

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);

Finalmente, use uma instância de contêiner para chamar executeCosmosBatch com o lote:

CosmosBatchResponse response = container.executeCosmosBatch(batch);

Assim que a resposta for recebida, verifique se a resposta é bem-sucedida. Se a resposta indicar um sucesso, extraia os resultados.

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

Importante

Se houver uma falha, a operação com falha terá um código de status de seu erro correspondente. Todas as outras operações terão um código de status 424 (dependência falhada). Se a operação falhar porque tenta criar um item que já existe, será retornado um código de estado 409 (conflito). O código de estado permite identificar a causa da falha da transação.

Píton

Obter ou criar uma instância de contêiner:

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

Em Python, as operações de Lote Transacional são muito semelhantes às operações singulares 'apis', e são tuplas contendo (operation_type_string, args_tuple, batch_operation_kwargs_dictionary). Abaixo encontram-se itens de exemplo que serão utilizados para demonstrar a funcionalidade de operações em lote.

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:
# /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"
}

Prepare as operações para serem adicionadas ao lote.

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")})

Adicione as operações ao lote:

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

Finalmente, execute o lote:

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]

Nota para usar a operação de correção e a operação de substituir_se_corresponder_etag no lote
O dicionário kwargs de operação em lote é limitado e aceita apenas um total de três valores de chave diferentes. No caso de se querer utilizar o "conditional patching" dentro do lote, a utilização da chave "filter_predicate" está disponível para a operação de patch. No caso de se pretender utilizar etags com qualquer uma das operações, as chaves "if_match_etag"/"if_none_match_etag" estão igualmente disponíveis.

batch_operations = [
       ("replace", (item_id, item_body), {"if_match_etag": etag}),
       ("patch", (item_id, operations), {"filter_predicate": filter_predicate, "if_none_match_etag": etag}),
   ]

Importante

Se houver uma falha, a operação com falha terá um código de status de seu erro correspondente. Todas as outras operações terão um código de status 424 (dependência falhada). Se a operação falhar porque tenta criar um item que já existe, será retornado um código de estado 409 (conflito). O código de estado permite identificar a causa da falha da transação.

Vá

Ao criar uma operação de lote transacional, chame NewTransactionalBatch.

pk := azcosmos.NewPartitionKeyString("road-bikes")

batch := container.NewTransactionalBatch(pk)

Em seguida, adicione várias operações ao lote:

type Bike struct {
	ID       string `json:"id"`
	Category string `json:"category"`
	Name     string `json:"name"`
}

bike := Bike{ID: "68719520766", Category: "road-bikes", Name: "Chropen Road Bike"}
bikeItem, _ := json.Marshal(bike)

batch.CreateItem(bikeItem, nil)

type BikePart struct {
	ID        string `json:"id"`
	Category  string `json:"category"`
	Name      string `json:"name"`
	ProductID string `json:"productID"`
}

part := BikePart{ID: "68719519885", Category: "road-bikes", Name: "Tronosuros Tire", ProductID: bike.ID}
bikePartItem, _ := json.Marshal(part)

batch.CreateItem(bikePartItem, nil)

Finalmente, use uma instância de contêiner para chamar ExecuteTransactionalBatch com o lote:

resp, _ := container.ExecuteTransactionalBatch(context.Background(), batch, nil)

if resp.Success {
	// execute conditional logic
}

Importante

Se houver uma falha, a operação com falha terá um código de status de seu erro correspondente. Todas as outras operações terão um código de status 424 (dependência falhada). Se a operação falhar porque tenta criar um item que já existe, será retornado um código de estado 409 (conflito). O código de estado permite identificar a causa da falha da transação.

Como são executadas as operações transacionais em lote

Quando o Lote Transacional é executado, todas as operações no Lote Transacional são agrupadas, serializadas em uma única carga útil e enviadas como uma única solicitação para o serviço Azure Cosmos DB.

O serviço recebe o pedido e executa todas as operações dentro de um âmbito transacional, e devolve uma resposta usando o mesmo protocolo de serialização. Esta resposta é bem-sucedida ou fracassada e fornece respostas individuais de operação por operação.

O SDK expõe a resposta para você verificar o resultado e, opcionalmente, extrair cada um dos resultados das operações internas.

Limitações

Atualmente, existem dois limites conhecidos

• O limite de tamanho da solicitação do Azure Cosmos DB restringe o tamanho da carga útil do Lote Transacional para não exceder 2 MB e o tempo máximo de execução é de 5 segundos.
• Existe um limite atual de 100 operações por Lote Transacional para garantir que o desempenho esteja conforme o esperado e dentro dos SLAs.

Próximo passo

TransactionalBatch Utilização