Azure Cosmos DB verilerinde toplu işlemler gerçekleştirme

  • Şunlara uygulanır: ✅ NoSQL

Bu öğretici, Azure Cosmos DB Java V4 SDK'sında toplu işlemler gerçekleştirmenin yönergelerini sağlar. SDK'nın bu sürümü, yerleşik toplu yürütücü kitaplığıyla birlikte gelir. Java SDK'sının eski bir sürümünü kullanıyorsanız en son sürüme geçmeniz önerilir. Azure Cosmos DB Java V4 SDK'sı, Java toplu desteği için önerilen geçerli çözümdür.

Şu anda toplu yürütücü kitaplığı yalnızca NoSQL için Azure Cosmos DB ve Gremlin hesapları için API tarafından desteklenmektedir. Gremlin için API ile toplu yürütücü .NET kitaplığını kullanma hakkında bilgi edinmek için, Gremlin için Azure Cosmos DB'de toplu işlemler gerçekleştirme sayfasını inceleyin.

Prerequisites

  • Eğer bir Azure aboneliğiniz yoksa, başlamadan önce ücretsiz bir hesap oluşturun. İsterseniz uç noktası ile https://localhost:8081 de kullanabilirsiniz. Birincil Anahtar, Kimlik doğrulama istekleri bölümünde sağlanır.

  • Java Geliştirme Seti (JDK) 1.8+

    • Ubuntu’da JDK’yi yüklemek için apt-get install default-jdk komutunu çalıştırın.

    • JAVA_HOME ortam değişkenini JDK’nin yüklü olduğu klasöre işaret edecek şekilde ayarladığınızdan emin olun.

  • Bir Maven ikili arşivi indirin ve yükleyin

    • Ubuntu’da Maven’i yüklemek için apt-get install maven komutunu çalıştırabilirsiniz.

  • Java hızlı başlangıç makalesinin veritabanı hesabı oluşturma bölümünde açıklanan adımları kullanarak NoSQL için Azure Cosmos DB hesabı oluşturun.

Örnek uygulamayı kopyalama

Şimdi GitHub'dan Azure Cosmos DB için Java V4 SDK'sı için genel bir örnek deposu indirerek kodla çalışmaya geçelim. Bu örnek uygulamalar Azure Cosmos DB'de CRUD işlemleri ve diğer yaygın işlemleri gerçekleştirir. Depoyu kopyalamak için bir komut istemi açın, uygulamayı kopyalamak istediğiniz dizine gidin ve aşağıdaki komutu çalıştırın:

 git clone https://github.com/Azure-Samples/azure-cosmos-java-sql-api-samples

Kopyalanan depo, SampleBulkQuickStartAsync.java klasöründe bir örnek /azure-cosmos-java-sql-api-samples/tree/main/src/main/java/com/azure/cosmos/examples/bulk/async içerir. Uygulama, Azure Cosmos DB'de öğeleri toplu olarak oluşturmak, eklemek, değiştirmek ve silmek için belgeler oluşturur ve işlemleri yürütür. Sonraki bölümlerde örnek uygulamadaki kodu gözden geçireceğiz.

Azure Cosmos DB'de toplu yürütme

  1. Azure Cosmos DB'nin bağlantı dizesi bağımsız değişken olarak okunur ve /examples/common/AccountSettings.javadosyasında tanımlanan değişkenlere atanır. Bu ortam değişkenleri ayarlanmalıdır
ACCOUNT_HOST=your account hostname;ACCOUNT_KEY=your account primary key

Toplu örneği çalıştırmak için Main Sınıfını belirtin:

com.azure.cosmos.examples.bulk.async.SampleBulkQuickStartAsync
  1. CosmosAsyncClient Nesnesi aşağıdaki deyimler kullanılarak başlatılır:
client = new CosmosClientBuilder()
    .endpoint(AccountSettings.HOST)
        .credential(new DefaultAzureCredentialBuilder().build())
    .preferredRegions(preferredRegions)
    .contentResponseOnWriteEnabled(true)
    .consistencyLevel(ConsistencyLevel.SESSION).buildAsyncClient();
  1. Bu örnek, eşzamansız bir veritabanı ve kapsayıcı oluşturur. Daha sonra toplu işlemlerin yürütüleceği birden çok belge oluşturur. Bu belgeleri reaktif akış Flux<Family> nesnesine ekler:
Family andersenFamilyItem = Families.getAndersenFamilyItem();
Family wakefieldFamilyItem = Families.getWakefieldFamilyItem();
Family johnsonFamilyItem = Families.getJohnsonFamilyItem();
Family smithFamilyItem = Families.getSmithFamilyItem();

//  Setup family items to create
Flux<Family> families = Flux.just(andersenFamilyItem, wakefieldFamilyItem, johnsonFamilyItem, smithFamilyItem);
  1. Örnek, toplu oluşturma, güncelle ve ekleme (upsert), değiştirme ve silme yöntemlerini içerir. Her yöntemde BulkWriter Flux<Family> akışındaki aileleri, CosmosBulkOperations içindeki birden fazla yöntem çağrısına eşleriz. Bu işlemler başka bir reaktif akış nesnesine Flux<CosmosItemOperation>eklenir. Akış, başlangıçta oluşturduğumuz executeBulkOperations zaman uyumsuz container yöntemine aktarılır ve işlemler toplu olarak yürütülür. Örnek olarak aşağıdaki toplu oluşturma yöntemine bakın:
private void bulkCreateItems(Flux<Family> families) {
    Flux<CosmosItemOperation> cosmosItemOperations = families.map(
        family -> CosmosBulkOperations.getCreateItemOperation(family, new PartitionKey(family.getLastName())));
    container.executeBulkOperations(cosmosItemOperations).blockLast();
}
  1. Örnek uygulamayla aynı dizinde de bir sınıf BulkWriter.java vardır. Bu sınıf, toplu yürütme sırasında oluşabilecek hız sınırlama (429) ve zaman aşımı (408) hatalarının nasıl işleneceğini ve bu işlemlerin etkili bir şekilde yeniden denendiğini gösterir. Yerel ve genel aktarım hızı denetiminin nasıl uygulandığını da gösteren aşağıdaki yöntemlerde uygulanır.
private void bulkUpsertItemsWithBulkWriterAbstraction() {
    Family andersenFamilyItem = Families.getAndersenFamilyItem();
    Family wakefieldFamilyItem = Families.getWakefieldFamilyItem();
    CosmosItemOperation wakeFieldItemOperation = CosmosBulkOperations.getUpsertItemOperation(wakefieldFamilyItem, new PartitionKey(wakefieldFamilyItem.getLastName()));
    CosmosItemOperation andersonItemOperation = CosmosBulkOperations.getUpsertItemOperation(andersenFamilyItem, new PartitionKey(andersenFamilyItem.getLastName()));
    BulkWriter bulkWriter = new BulkWriter(container);
    bulkWriter.scheduleWrites(andersonItemOperation);
    bulkWriter.scheduleWrites(wakeFieldItemOperation);
    bulkWriter.execute().subscribe();
}

private void bulkUpsertItemsWithBulkWriterAbstractionAndLocalThroughPutControl() {
    ThroughputControlGroupConfig groupConfig =
            new ThroughputControlGroupConfigBuilder()
                    .setGroupName("group1")
                    .setTargetThroughput(200)
                    .build();
    container.enableLocalThroughputControlGroup(groupConfig);
    Family andersenFamilyItem = Families.getAndersenFamilyItem();
    Family wakefieldFamilyItem = Families.getWakefieldFamilyItem();
    CosmosItemOperation wakeFieldItemOperation = CosmosBulkOperations.getUpsertItemOperation(wakefieldFamilyItem, new PartitionKey(wakefieldFamilyItem.getLastName()));
    CosmosItemOperation andersonItemOperation = CosmosBulkOperations.getUpsertItemOperation(andersenFamilyItem, new PartitionKey(andersenFamilyItem.getLastName()));
    BulkWriter bulkWriter = new BulkWriter(container);
    bulkWriter.scheduleWrites(andersonItemOperation);
    bulkWriter.scheduleWrites(wakeFieldItemOperation);
    bulkWriter.execute().subscribe();
}

private void bulkCreateItemsWithBulkWriterAbstractionAndGlobalThroughputControl() {
    String controlContainerId = "throughputControlContainer";
    CosmosAsyncContainer controlContainer = database.getContainer(controlContainerId);
    database.createContainerIfNotExists(controlContainer.getId(), "/groupId").block();

    ThroughputControlGroupConfig groupConfig =
            new ThroughputControlGroupConfigBuilder()
                    .setGroupName("group-" + UUID.randomUUID())
                    .setTargetThroughput(200)
                    .build();

    GlobalThroughputControlConfig globalControlConfig = this.client.createGlobalThroughputControlConfigBuilder(this.database.getId(), controlContainerId)
            .setControlItemRenewInterval(Duration.ofSeconds(5))
            .setControlItemExpireInterval(Duration.ofSeconds(20))
            .build();

    container.enableGlobalThroughputControlGroup(groupConfig, globalControlConfig);
    CosmosItemRequestOptions requestOptions = new CosmosItemRequestOptions();
    requestOptions.setThroughputControlGroupName(groupConfig.getGroupName());
    Family andersenFamilyItem = Families.getAndersenFamilyItem();
    Family wakefieldFamilyItem = Families.getWakefieldFamilyItem();
    CosmosItemOperation andersonItemOperation = CosmosBulkOperations.getCreateItemOperation(andersenFamilyItem, new PartitionKey(andersenFamilyItem.getLastName()));
    CosmosItemOperation wakeFieldItemOperation = CosmosBulkOperations.getCreateItemOperation(wakefieldFamilyItem, new PartitionKey(wakefieldFamilyItem.getLastName()));
    BulkWriter bulkWriter = new BulkWriter(container);
    bulkWriter.scheduleWrites(andersonItemOperation);
    bulkWriter.scheduleWrites(wakeFieldItemOperation);
    bulkWriter.execute().subscribe();
}
  1. Ayrıca, örnekte yanıt işleme ekleme ve yürütme seçeneklerini ayarlama işlemlerini gösteren toplu oluşturma yöntemleri vardır:
private void bulkCreateItemsWithResponseProcessing(Flux<Family> families) {
    Flux<CosmosItemOperation> cosmosItemOperations = families.map(
        family -> CosmosBulkOperations.getCreateItemOperation(family, new PartitionKey(family.getLastName())));
    container.executeBulkOperations(cosmosItemOperations).flatMap(cosmosBulkOperationResponse -> {

        CosmosBulkItemResponse cosmosBulkItemResponse = cosmosBulkOperationResponse.getResponse();
        CosmosItemOperation cosmosItemOperation = cosmosBulkOperationResponse.getOperation();

        if (cosmosBulkOperationResponse.getException() != null) {
            logger.error("Bulk operation failed", cosmosBulkOperationResponse.getException());
        } else if (cosmosBulkItemResponse == null ||
            !cosmosBulkOperationResponse.getResponse().isSuccessStatusCode()) {

            logger.error(
                "The operation for Item ID: [{}]  Item PartitionKey Value: [{}] did not complete " +
                    "successfully with " + "a" + " {} response code.",
                cosmosItemOperation.<Family>getItem().getId(),
                cosmosItemOperation.<Family>getItem().getLastName(),
                cosmosBulkItemResponse != null ? cosmosBulkItemResponse.getStatusCode() : "n/a");
        } else {
            logger.info(
                "Item ID: [{}]  Item PartitionKey Value: [{}]",
                cosmosItemOperation.<Family>getItem().getId(),
                cosmosItemOperation.<Family>getItem().getLastName());
            logger.info("Status Code: {}", cosmosBulkItemResponse.getStatusCode());
            logger.info("Request Charge: {}", cosmosBulkItemResponse.getRequestCharge());
        }
        if (cosmosBulkItemResponse == null) {
            return Mono.error(new IllegalStateException("No response retrieved."));
        } else {
            return Mono.just(cosmosBulkItemResponse);
        }
    }).blockLast();
}

private void bulkCreateItemsWithExecutionOptions(Flux<Family> families) {
    CosmosBulkExecutionOptions bulkExecutionOptions = new CosmosBulkExecutionOptions();

    // The default value for maxMicroBatchConcurrency is 1.
    // By increasing it, it means more concurrent requests will be allowed to be sent to the server, which leads to increased RU usage.
    //
    // Before you increase the value, please examine the RU usage of your container - whether it has been saturated or not.
    // When the RU has already been under saturation, increasing the concurrency will not help the situation,
    // rather it may cause more 429 and request timeout.
    bulkExecutionOptions.setMaxMicroBatchConcurrency(2);
    Flux<CosmosItemOperation> cosmosItemOperations = families.map(family -> CosmosBulkOperations.getCreateItemOperation(family, new PartitionKey(family.getLastName())));
    container.executeBulkOperations(cosmosItemOperations, bulkExecutionOptions).blockLast();
}

Performans ipuçları

Toplu yürütücü kütüphanesini kullanırken daha iyi performans sağlamak için aşağıdaki noktaları dikkate alın:

  • En iyi performans için uygulamanızı Azure Cosmos DB hesabınızın yazma bölgesiyle aynı bölgedeki bir Azure VM'sinden çalıştırın.

  • Daha yüksek aktarım hızı elde etmek için:

    • Çok sayıda belgeyi işlerken herhangi bir bellek sorununu önlemek için JVM'nin yığın boyutunu yeterince büyük bir sayıya ayarlayın. Önerilen yığın boyutu: max(3 GB, 3 * sizeof(tek bir toplu işlemde toplu içeri aktarma API'sine geçirilen tüm belgeler)).
    • Çok sayıda belgeyle toplu işlemler gerçekleştirirken daha yüksek aktarım hızı elde ettiğiniz için ön işleme süresi vardır. Bu nedenle, 10.000.000 belgeyi içeri aktarmak istiyorsanız, her biri 1.000.000 belge boyutunda 10 belge topluluğunda 10 kez toplu içeri aktarma çalıştırmak, her biri 100.000 belge boyutunda 100 belge topluluğunda toplu içeri aktarma çalıştırmaktan daha iyidir.

  • Belirli bir Azure Cosmos DB kapsayıcısına karşılık gelen tek bir sanal makine içinde uygulamanın tamamı için tek bir CosmosAsyncClient nesnesi örneği oluşturmanız önerilir.

  • Tek bir toplu işlem API'sinin yürütülmesi, istemci makinesinin CPU'su ve ağ G/Ç'sinin büyük bir bölümünü tüketir. Bu durum, dahili olarak birden çok görev ortaya çıkararak gerçekleşir, toplu işlem API çağrılarını yürüten her uygulama işleminizde birden çok eşzamanlı görev oluşturmaktan kaçının. Tek bir sanal makinede çalıştırılan tek bir toplu işlem API'si çağrıları kapsayıcınızın aktarım hızının tamamını tüketemiyorsa (kapsayıcınızın aktarım hızı > 1 milyon RU/sn ise), toplu işlem API çağrılarını eşzamanlı olarak yürütmek için ayrı sanal makineler oluşturmak tercih edilir.

Sonraki Adımlar

  • Last updated on