使用Javascript SDK时，如何在Amazon Kinesis consumer中获得最新记录？

Question

我正在使用AWS Javascript SDK从Kinesis数据流中消费。我想获取碎片中的最新记录。

当我将ShardIterator Type设置为“最新”时，我不会取回任何记录。但是，当我使用“TRIM_HORIZON”时，我会取回所有记录。

kinesis.describeStream(describeParams, function(err, data) {
            if (err) {
                console.log(err, err.stack);    // an error occurred
            }
            else {
                var getParams = {
                    ShardId: data.StreamDescription.Shards[0].ShardId,
                    ShardIteratorType: "TRIM_HORIZON",     //get oldest package
                    StreamName: streamName,
                };

                if(shardIteratorType){
                    console.log("you have passed some shardIteratorType:" + shardIteratorType);
                    getParams.ShardIteratorType = shardIteratorType;
                }

                kinesis.getShardIterator(getParams, function(err, result) {
                    if (err) {
                        console.log("Error in getShardIterator()");
                        console.log(err);
                    } else {
                        console.log("calling getRecord with shard iterator");
                        // Get records from the Kinesis stream
                        getRecord(result.ShardIterator);
                    }
                });
            }
        });

function getRecord(shard_iterator) {
    console.log("getRecord was called.");
    var getRecParams = {
        ShardIterator: shard_iterator
    };

    kinesis.getRecords(getRecParams, function(err, result) {
        if (err) {
            console.log("Error in getRecords() from the Kinesis stream.");
            console.log(err);
        } else {
            try {

            if(result.Records.length > 0) {
                    // Loop through all the packages
               for(var i = 0; i < result.Records.length; i++) {
                   if(result.Records[i] != undefined) {
                        var getData = JSON.parse( decodeURIComponent
(escape(result.Records[i].Data)));
                        console.log(getData);
                        var table = document.getElementById("myTable");
                        var row = table.insertRow(0);
                        var j = i + 1 ;
                        var cell1 = row.insertCell(0);


                        cell1.innerHTML = getData ;

                    }
                }
            }
            } catch(err) {
                console.log("Error parsing the package.");
                console.log(err);
            }
        }
    });
}

对于shard迭代器类型，我希望使用“latest”而不是“TRIM_HORIZON”来获得最新的记录，而不是所有的历史记录。我做错了什么？

Answer 1

来自文件https://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetShardIterator.html

在请求中，您可以指定分片迭代器类型AT_TIMESTAMP从任意时间点读取记录，TRIM_HORIZON使ShardIterator指向系统分片中最后一条未修剪的记录（分片中最旧的数据记录），或LATEST，以便您始终读取分片中的最新数据。

这里的最新意味着“现在”跳过最近检查点和现在之间的所有记录。

使用LATEST作为shard迭代器类型，可以认为它读取了在使用getShardIterator函数返回的shard迭代器之后立即出现的记录。

您可以在下面参考它如何与“最新”一起工作的示例：

kinesis.describeStream(describeParams, (err, streamData) => {
    // Skipping error handling

    kinesis.getShardIterator(getShardIteratorParams, (err, shardIterData) => {
        let shardIterator = shardIterData.ShardIterator;

        // Keep reading records from the stream
        while (true) {
            let getRecParams = {
                ShardIterator: shardIterator
            };
            kinesis.getRecords(getRecParams, (err, recData) => {
                // Skipping error handling

                if (recData.Records.length > 0) {
                   // Do something

                   shardIterator = recData.NextShardIterator;
                }
            });
            // Break if you need
        }
    });
});