DynamoDB一批執行QueryRequests

Question

我有以下DynamoDB查詢與哈希蘋果和時間戳小於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一樣。

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

Answer 1

從AWS論壇here一個非常類似的問題：

DynamoDB的查詢API僅支持查詢操作的上述指標的單一的「使用」，並作爲一個結果，「散列「你正在查詢的索引必須被指定爲EQ條件。 DynamoDB目前沒有任何種類的「批量查詢」API，因此不幸的是您今天在單個API調用中無法找到您想要的內容。如果這些是GetItem請求（不適合您的用例），您可以發出一個BatchGetItem請求。

與此同時，由於它看起來像你正在使用Java，我的建議是使用線程並行發出多個查詢請求。下面是實現這一一些示例代碼，但你要考慮你希望你的應用程序來處理分頁/部分結果和錯誤信息：

/** 
* 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); 
    } 
}