DynamoDB批量执行QueryRequests [英] DynamoDB batch execute QueryRequests

问题描述

我有以下DynamoDB查询，该查询返回的第一条记录的哈希为 apple ，且时间戳小于 some_timestamp 。

I have the following DynamoDB query which returns the first record with the hash apple and time-stamp less than some_timestamp.

Map<String, Condition> keyConditions = newHashMap();

keyConditions.put("HASH", new Condition().
    withComparisonOperator(EQ).
    withAttributeValueList(new AttributeValue().withS("apple")))
);

keyConditions.put("TIMESTAMP", new Condition().
    withComparisonOperator(LE).
    withAttributeValueList(new AttributeValue().withN(some_timestamp)))
);

QueryResult queryResult = dynamoDBClient.query(
    new QueryRequest().
            withTableName("TABLE").
            withKeyConditions(keyConditions).
            withLimit(1).
            withScanIndexForward(SCAN_INDEX_FORWARD)
);

我需要执行许多此类查询，所以我的问题是：是否可以批量执行这些查询查询？类似于以下API。

I need to execute many queries of this kind and so my question: is it possible to batch execute these queries? Something like the following API.

Map<String, Condition> keyConditions = newHashMap();

keyConditions.put("HASH", new Condition().
    withComparisonOperator(EQ).
    withAttributeValueList(new AttributeValue().withS("apple")))
);

keyConditions.put("TIMESTAMP", new Condition().
    withComparisonOperator(LE).
    withAttributeValueList(new AttributeValue().withN(some_timestamp)))
);

QueryRequest one = new QueryRequest().
    withTableName("TABLE").
    withKeyConditions(keyConditions).
    withLimit(1).
    withScanIndexForward(SCAN_INDEX_FORWARD);

keyConditions = newHashMap();

keyConditions.put("HASH", new Condition().
    withComparisonOperator(EQ).
    withAttributeValueList(new AttributeValue().withS("pear")))
);

keyConditions.put("TIMESTAMP", new Condition().
    withComparisonOperator(LE).
    withAttributeValueList(new AttributeValue().withN(some_other_timestamp)))
);

QueryRequest two = new QueryRequest().
    withTableName("TABLE").
    withKeyConditions(keyConditions).
    withLimit(1).
    withScanIndexForward(SCAN_INDEX_FORWARD)

ArrayList<String> queryRequests = new ArrayList<String>() {{
    add(one);
    add(two);
}};

List<QueryResult> queryResults = dynamoDBClient.query(queryRequests);

推荐答案

来自AWS论坛中一个非常类似的问题此处：

From a very similar question in the AWS forums here:

DynamoDB的查询API在查询操作中仅支持索引的单个使用，因此，必须将要查询的索引的哈希指定为EQ条件。 DynamoDB当前没有任何类型的批处理查询 API，因此，遗憾的是，今天您无法通过单个API调用来寻找所需的内容。如果这些是GetItem请求（虽然不适合您的用例），则可以发出BatchGetItem请求。

DynamoDB's Query API only supports a single "use" of the index in the query operation, and as a result, the "hash" of the index you're querying has to be specified as an EQ condition. DynamoDB does not currently have any kind of "batch query" API, so unfortunately what you're looking for is not possible today in a single API call. If these were GetItem requests (not suitable for your use case though), you could issue a BatchGetItem request.

同时，因为看起来您正在使用Java ，我的建议是使用线程并行发出多个查询请求。这是一些完成此操作的示例代码，但是您需要考虑如何让应用程序处理分页/部分结果以及错误：

In the meantime, since it looks like you're using Java, my recommendation would be to use threads to issue multiple query requests in parallel. Here's some sample code that accomplishes this, but you'll want to consider how you want your application to handle pagination / partial results, and errors:

/**
* Simulate a "Batch Query" operation in DynamoDB by querying an index for 
* multiple hash keys
* 
* Resulting list may be incomplete if any queries time out.  Returns a list of 
* QueryResult so that LastEvaluatedKeys can be followed.  A better implementation 
* would answer the case where some queries fail, deal with pagination (and 
* Limit), have configurable timeouts.  One improvement on this end would be 
* to make a simple immutable bean that contains a query result or exception, 
* as well as the associated request.  Maybe it could even be called back with 
* a previous list for pagination. 
* 
* @param hashKeyValues (you'll also need table name / index name) 
* @return a list of query results for the queries that succeeded
* @throws InterruptedException
*/
public List<QueryResult> queryAll(String... hashKeyValues) 
  throws InterruptedException {
  // initialize accordingly
  int timeout = 2 * 1000;
  ExecutorService executorService = Executors.newFixedThreadPool(10);

  final List<QueryResult> results = 
    new ArrayList<QueryResult>(hashKeyValues.length);
  final CountDownLatch latch = 
    new CountDownLatch(hashKeyValues.length);

  // Loop through the hash key values to "OR" in the final list of results
  for (final String hashKey : hashKeyValues) {

    executorService.submit(new Runnable() {

      @Override
      public void run() {
        try {
          // fill in parameters
          QueryResult result = dynamodb.query(new QueryRequest()
            .withTableName("MultiQueryExample")
            .addKeyConditionsEntry("City", new Condition()
              .withComparisonOperator("EQ")
            .withAttributeValueList(new AttributeValue(hashKey))));
          // one of many flavors of dealing with concurrency
          synchronized (results) { 
            results.add(result);
          }
        } catch (Throwable t) {
          // Log and handle errors
          t.printStackTrace();
        } finally {
          latch.countDown();
        }
      }
    });
  }

  // Wait for all queries to finish or time out
  latch.await(timeout, TimeUnit.MILLISECONDS);

  // return a copy to prevent concurrent modification of 
  // the list in the face of timeouts
  synchronized (results) {
    return new ArrayList<QueryResult>(results);
  }
}

这篇关于DynamoDB批量执行QueryRequests的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！