DynamoDb 流，刚刚获得新的更新

Question

我正在尝试使用 DynamoDb 流，我正在使用this article 中显示的示例代码。我已经修改它以在基本的 Spring Boot 应用程序 (initializr) 中工作，利用启用了流的现有 DynamoDb 表。然而，一切似乎都可以工作；我没有看到任何新的更新。

此特定数据库每天在特定时间进行一次批量更新，它可能会在一天中不时地进行一些细微的更改。我正在尝试监视这些次要更新。当我运行应用程序时，我可以看到批量更新的记录，但是如果我的应用程序正在运行并且我使用 AWS 控制台修改、创建或删除记录，我似乎没有得到任何输出。

我正在使用：

• Spring Boot:2.3.9.RELEASE
• amazon-kinesis-client:1.14.2
• Java 11
• 在 Mac Catalina 上运行（尽管这无关紧要）

在我的测试应用程序中，我执行了以下操作：

package com.test.dynamodb_streams_test_kcl.service;

import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBStreams;
import com.amazonaws.services.dynamodbv2.model.*;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.util.List;

@Slf4j
@Service
@RequiredArgsConstructor
public class LowLevelKclProcessor {
    private static final String dynamoDbTableName = "global-items";

    private final AmazonDynamoDB dynamoDB;
    private final AmazonDynamoDBStreams dynamoDBStreams;
    private final ZonedDateTime startTime = ZonedDateTime.now();

    @PostConstruct
    public void initialize() {
        log.info("Describing table={}", dynamoDbTableName);
        DescribeTableResult itemTableDescription = dynamoDB.describeTable(dynamoDbTableName);
        log.info("Got description");
        String itemTableStreamArn = itemTableDescription.getTable().getLatestStreamArn();
        log.info("Got stream arn ({}) for table={} tableArn={}", itemTableStreamArn,
                itemTableDescription.getTable().getTableName(), itemTableDescription.getTable().getTableArn());

        // Get all the shard IDs from the stream.  Note that DescribeStream returns
        // the shard IDs one page at a time.
        String lastEvaluatedShardId = null;

        do {
            DescribeStreamResult describeStreamResult = dynamoDBStreams.describeStream(
                    new DescribeStreamRequest()
                            .withStreamArn(itemTableStreamArn)
                            .withExclusiveStartShardId(lastEvaluatedShardId));
            List<Shard> shards = describeStreamResult.getStreamDescription().getShards();

            // Process each shard on this page

            for (Shard shard : shards) {

                String shardId = shard.getShardId();
                System.out.println("Shard: " + shard);

                // Get an iterator for the current shard

                GetShardIteratorRequest getShardIteratorRequest = new GetShardIteratorRequest()
                        .withStreamArn(itemTableStreamArn)
                        .withShardId(shardId)
                        .withShardIteratorType(ShardIteratorType.LATEST);
                GetShardIteratorResult getShardIteratorResult =
                        dynamoDBStreams.getShardIterator(getShardIteratorRequest);
                String currentShardIter = getShardIteratorResult.getShardIterator();

                // Shard iterator is not null until the Shard is sealed (marked as READ_ONLY).
                // To prevent running the loop until the Shard is sealed, which will be on average
                // 4 hours, we process only the items that were written into DynamoDB and then exit.
                int processedRecordCount = 0;
                while (currentShardIter != null && processedRecordCount < 100) {
                    System.out.println("    Shard iterator: " + currentShardIter.substring(380));

                    // Use the shard iterator to read the stream records

                    GetRecordsResult getRecordsResult = dynamoDBStreams.getRecords(new GetRecordsRequest()
                            .withShardIterator(currentShardIter));
                    List<Record> records = getRecordsResult.getRecords();
                    for (Record record : records) {
                        // I set a breakpoint on the line below, but it was never hit after the bulk update info
                        if (startTime.isBefore(ZonedDateTime.ofInstant(record.getDynamodb()
                                .getApproximateCreationDateTime().toInstant(), ZoneId.systemDefault()))) {
                            System.out.println("        " + record.getDynamodb());
                        }
                    }
                    processedRecordCount += records.size();
                    currentShardIter = getRecordsResult.getNextShardIterator();
                }
            }

            // If LastEvaluatedShardId is set, then there is
            // at least one more page of shard IDs to retrieve
            lastEvaluatedShardId = describeStreamResult.getStreamDescription().getLastEvaluatedShardId();

        } while (lastEvaluatedShardId != null);
    }
}

Answer 1

注意，您的测试基于低级 API，而不是 Kenisis 客户端库。所以处理一些棘手的技术细节是很正常的。

---

您的测试应用程序与doc 中给出的示例有一些相似之处，但存在问题：

当我运行应用程序时，我可以看到批量更新的记录

ShardIteratorType.LATEST 不会查找运行测试之前发生的旧记录（它会在分片中最近的流记录之后开始读取）

因此，我将假设迭代器类型不同（例如：TRIM_HORIZON）并在您的测试过程中稍后更改为 LATEST。

主要问题在于您的应用程序将顺序轮询分片，并且它将在第一个分片中阻塞，直到在该分片中找到 100 条新记录（由于 LATEST 迭代器类型）。

因此，如果新的细微更改属于不同的分片，则在测试运行时您可能看不到它们。

解决方案：

1- 使用线程并行轮询分片。

2- 使用最后记录的记录的序列号过滤返回的分片，并尝试猜测可能包含微小变化的分片。

3- 危险，我不确定它是否有效 :) 在测试表中，如果您的数据模型允许：关闭当前流，并启用一个新流，然后确保所有写入都属于一个分区。在大多数情况下，表分区与活动分片具有一对一的关系。从理论上讲，您只有一个活动分片要处理。