{"id":3848,"date":"2022-01-20T06:00:14","date_gmt":"2022-01-20T14:00:14","guid":{"rendered":"https:\/\/devblogs.microsoft.com\/cosmosdb\/?p=3848"},"modified":"2022-01-18T11:29:19","modified_gmt":"2022-01-18T19:29:19","slug":"read-many-items-fast-with-the-java-sdk-for-azure-cosmos-db","status":"publish","type":"post","link":"https:\/\/devblogs.microsoft.com\/cosmosdb\/read-many-items-fast-with-the-java-sdk-for-azure-cosmos-db\/","title":{"rendered":"Read many items fast with the Java SDK for Azure Cosmos DB"},"content":{"rendered":"

Typically, when retrieving many records from a database, you will run filter or aggregation queries. In these scenarios, latency is usually less important than the accuracy of the results. However, there are some use cases where many separate records need to be collected back very quickly. For example, you may be building a comparison site where thousands of entities are maintained independently, but frequently need to be pulled back during user comparisons.<\/p>\n

In this situation, you can have a trade-off between the simplicity (but high latency) of running a large query, or the lower latency (but high client-side compute cost and complexity) of hitting the database with many point reads<\/a> in parallel.<\/p>\n

The following has been extracted from a real-world example, and has been codified into a more generic sample here.<\/a> The sample demonstrates how the readMany<\/a><\/span> method in Azure Cosmos DB Java SDK<\/a> can help in this scenario. In this case we are using the readMany<\/a><\/span> method along with CosmosAsyncContainer<\/a> in a multi-threaded application using the reactor<\/a> framework (note: the Azure Cosmos DB .NET SDK<\/a> has the same method<\/a>, which can be used in the same way).<\/p>\n

Suppose you have a client machine (or machines), and you want to bring back 1,000 records using point reads as quickly as possible. In our demo sample,<\/a> we have two methods that are compared for differences in their performance.<\/p>\n

First, we have a method that executes point reads using as many concurrent threads as possible on the local client machine:<\/p>\n

\n
\n
\n
\u00a0 \u00a0 private void readManyItems() throws InterruptedException {\r\n\u00a0 \u00a0 \u00a0 \u00a0 \/\/ collect the ids that were generated when writing the data.\r\n\u00a0 \u00a0 \u00a0 \u00a0 List<String> list = new ArrayList<String>();\r\n\u00a0 \u00a0 \u00a0 \u00a0 for (final JsonNode doc : docs) {\r\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 list.add(doc.get(\"id\").asText());\r\n\u00a0 \u00a0 \u00a0 \u00a0 }\r\n\u00a0 \u00a0 \u00a0 \u00a0 final long startTime = System.currentTimeMillis();\r\n\u00a0 \u00a0 \u00a0 \u00a0 Flux.fromIterable(list)\r\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 .flatMap(id -> container.readItem(id, new PartitionKey(id), Item.class))\r\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 .flatMap(itemResponse -> {\r\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 if (itemResponse.getStatusCode() == 200) {\r\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 double requestCharge = itemResponse.getRequestCharge();\r\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 BinaryOperator<Double> add = (u, v) -> u + v;\r\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 totalRequestCharges.getAndAccumulate(requestCharge, add);\r\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 } else\r\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 logger.info(\"WARNING insert status code {} != 200\" + itemResponse.getStatusCode());\r\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 request_count.getAndIncrement();\r\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 return Mono.empty();\r\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 }).subscribe();\r\n\u00a0 \u00a0 \u00a0 \u00a0 logger.info(\"Waiting while subscribed async operation completes all threads...\");\r\n\u00a0 \u00a0 \u00a0 \u00a0 while (request_count.get() < NUMBER_OF_DOCS) {\r\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \/\/ looping while subscribed async operation completes all threads\r\n\u00a0 \u00a0 \u00a0 \u00a0 }\r\n\u00a0 \u00a0 \u00a0 \u00a0 if (request_count.get() == NUMBER_OF_DOCS) {\r\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 request_count.set(0);\r\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 final long endTime = System.currentTimeMillis();\r\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 final long duration = (endTime - startTime);\r\n\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 totalReadLatency.getAndAdd(duration);\r\n\u00a0 \u00a0 \u00a0 \u00a0 }\r\n\u00a0 \u00a0 }<\/pre>\n<\/div>\n<\/div>\n<\/div>\n
<\/div>\n
\n
Next, we have a method that divides the requests into chunks using the NUMBER_OF_DOCS_PER_THREAD parameter, and each thread will send a list of id and partition key tuples to the database to be executed using readMany<\/a><\/span>:<\/p>\n
    private void readManyItemsEnhanced() throws InterruptedException {\r\n        \/\/ collect the ids that were generated when writing the data.\r\n        List<String> list = new ArrayList<String>();\r\n        for (final JsonNode doc : docs) {\r\n            list.add(doc.get(\"id\").asText());\r\n        }\r\n        List<List<String>> lists = ListUtils.partition(list, NUMBER_OF_DOCS_PER_THREAD);\r\n\r\n        final long startTime = System.currentTimeMillis();\r\n        Flux.fromIterable(lists).flatMap(x -> {\r\n\r\n            List<CosmosItemIdentity> pairList = new ArrayList<>();\r\n\r\n            \/\/ add point reads in this thread as a list to be sent to Cosmos DB\r\n            for (final String id : x) {\r\n                \/\/ increment request count here so that total requests will equal total docs\r\n                request_count.getAndIncrement();\r\n                pairList.add(new CosmosItemIdentity(new PartitionKey(String.valueOf(id)), String.valueOf(id)));\r\n            }\r\n\r\n            \/\/ instead of reading sequentially, send CosmosItem id and partition key tuple of items to be read\r\n            Mono<FeedResponse<Item>> documentFeedResponse = container.readMany(pairList, Item.class);                    \r\n            double requestCharge = documentFeedResponse.block().getRequestCharge();\r\n            BinaryOperator<Double> add = (u, v) -> u + v;\r\n            totalEnhancedRequestCharges.getAndAccumulate(requestCharge, add);\r\n            return documentFeedResponse;\r\n        })\r\n                .subscribe();\r\n\r\n        logger.info(\"Waiting while subscribed async operation completes all threads...\");\r\n        while (request_count.get() < NUMBER_OF_DOCS) {\r\n            \/\/ looping while subscribed async operation completes all threads\r\n        }\r\n        if (request_count.get() == NUMBER_OF_DOCS) {\r\n            request_count.set(0);\r\n            final long endTime = System.currentTimeMillis();\r\n            final long duration = (endTime - startTime);\r\n            totalEnhancedReadLatency.getAndAdd(duration);\r\n        }\r\n    }<\/pre>\n<\/div>\n
<\/div>\n
First, let’s see what happens when we run the demo and configure just 10 documents per thread for the enhanced readManyItemsEnhanced() method:<\/span><\/div>\n
<\/code><\/code><\/div>\n
\n
public static final int NUMBER_OF_DOCS = 1000;\r\npublic static final int NUMBER_OF_DOCS_PER_THREAD = 10;\r\n\r\nINFO: Reading many items.... \r\nINFO: Waiting while subscribed async operation completes all threads...\r\nINFO: Reading many items (enhanced using readMany method)\r\nINFO: Waiting while subscribed async operation completes all threads...\r\nINFO: Total latency with standard multi-threading: 907 \r\nINFO: Total latency using readMany method: 2109 \r\nINFO: Total request charges with standard multi-threading: 1000.0 \r\nINFO: Total request charges using readMany method: 927.1299999999991\r\n\r\n<\/pre>\n<\/div>\n
\n
 <\/p>\n
In this test, we can see that latency is higher when using readMany<\/a><\/span>. With the local resources available, executing point reads with standard multi-threading performs better than splitting the requests and sending 10 requests per thread using readMany<\/a><\/span>. However, notice that the total request unit<\/a> charges are slightly lower with readMany<\/a><\/span>. Now let’s see what happens when we increase the number of NUMBER_OF_DOCS_PER_THREAD to 100:<\/p>\n
\n
\n
public static final int NUMBER_OF_DOCS = 1000;\r\npublic static final int NUMBER_OF_DOCS_PER_THREAD = 100;\r\n\r\nINFO: Reading many items.... \r\nINFO: Waiting while subscribed async operation completes all threads...\r\nINFO: Reading many items (enhanced using readMany method)\r\nINFO: Waiting while subscribed async operation completes all threads...\r\nINFO: Total latency with standard multi-threading: 825\r\nINFO: Total latency using readMany method: 380\r\nINFO: Total request charges with standard multi-threading: 1000.0\r\nINFO: Total request charges using readMany method: 212.79000000000002<\/pre>\n
<\/code><\/p>\n<\/div>\n<\/div>\n<\/div>\n
<\/div>\n
\n
We can now see that latency is significantly reduced compared to using multi-threading. Not only that, but RU charges are relatively much lower for the same number of point reads! This approach is a great solution if you need to run many point reads in parallel, but don’t have a large number of client machines or cores standing by to serve a large “scatter gather” request, and need to mitigate the trade-off on overall latency. Keep in mind that the extent to which this approach will benefit latency in the application will depend on the average size of each document, so you should benchmark the performance when implementing this.<\/p>\n
Get Started<\/i><\/h3>\n