AWS Hive + EMR 上的 Kinesis = 了解检查点

Question

我有一个 AWS Kinesis 流，并在 Hive 中创建了一个指向它的外部表。然后，我为检查点创建一个 DynamoDB 表，并在我的 Hive 查询中设置以下属性，如 here:

所述

set kinesis.checkpoint.enabled=true;
set kinesis.checkpoint.metastore.table.name=my_dynamodb_table;
set kinesis.checkpoint.metastore.hash.key.name=HashKey;                                                               
set kinesis.checkpoint.metastore.range.key.name=RangeKey;                                                            
set kinesis.checkpoint.logical.name=my_logical_name;                                                                 
set kinesis.checkpoint.iteration.no=0;

我有以下问题：

• 我是否必须始终将 iteration.no 设置为 0？
• 这是否总是从脚本的开头开始（即将被驱逐的最旧的 Kinesis 记录）？
• 假设我设置了一个 cron 来安排脚本的执行，我如何检索“下一个”迭代次数？
• 要在相同的数据上重新执行脚本，以相同的执行号重新运行查询是否足够？
• 如果我一遍又一遍地执行 select * from kinesis_ext_table limit 100 和 iteration.no=0，一旦第一个 Kinesis 记录开始被驱逐，我会得到不同/奇怪的结果吗？

鉴于 DynamoDB 检查点条目：

{"startSeqNo":"1234",
 "endSeqNo":"5678",
 "closed":false}

• closed 字段的含义是什么？
• 序列号是否递增，开始和结束之间是否存在关系（例如：结束 - 开始 = 读取的记录数）？
• 我注意到有时只有 endSeqNum（没有 startSeqNum），我该如何解释？

我知道有很多问题，但我在文档中找不到这些答案。

Answer 1

查看Kinesis documentation 和Kinesis Storage Handler Readme，其中包含您许多问题的答案。

我是否总是必须从 iteration.no 设置为 0 开始？

是的，除非您正在执行一些高级逻辑，需要您跳过流中已知或已处理的部分

这是否总是从脚本的开头开始（最旧的 Kinesis 记录即将被驱逐）？

是的

假设我设置了一个 cron 来安排脚本的执行，我如何检索“下一个”迭代次数？

这由 hive 脚本处理，因为它在每次运行时都会查询 kinesis 流中的所有数据

要对相同的数据重新执行脚本，以相同的执行次数重新运行查询是否足够？

由于 Kinesis 数据是一个 24 小时的时间窗口，因此自您上次查询以来数据（可能）已更改，因此您可能希望在 Hive 作业中再次查询所有记录

如果我一遍又一遍地执行 select * from kinesis_ext_table limit 100with iteration.no=0，一旦第一个 Kinesis 记录开始被驱逐，我会得到不同/奇怪的结果吗？

是的，您希望结果会随着流的变化而变化

鉴于 DynamoDB 检查点条目： 闭场是什么意思？

虽然这是 Kinesis Storage Handler 的内部细节，但我相信这表明分片是否是父分片，这表明它是打开并接受新数据还是关闭且不接受新数据进入分片。如果您已向上或向下扩展您的流，父分片会存在 24 小时，并包含自您扩展后的所有数据，但不会将新数据插入这些分片。

序列号是递增的吗，开始和结束之间是否存在关系（EG：结束-开始=读取的记录数）？

新的序列号通常会随着时间的推移而增加，这是亚马逊在这方面提供的唯一指导。

我注意到有时只有 endSeqNum（没有 startSeqNum），我应该如何解释？

这意味着分片已打开并且仍在接受新数据（不是父分片）