问题标签 [amazon-kinesis]

我想通过云形成模板创建 Kinesis 资源，但它不允许我提供“StreamName”作为资源的属性。

它说“无法识别的属性“StreamName”。如何在我的模板中提供流名称。谢谢，Nithya。

我正在运行一个脚本来抓取推文并将它们放入 Kinesis 流中。如果我停止脚本运行并等待 5-10 分钟重新启动它，它似乎接受新数据并在没有错误的响应中返回 ShardID 和 SequenceNumber。

当我的消费者脚本尝试查找/解析/使用新添加的数据时，它无处可寻，但在停止脚本之前的任何数据仍然可以使用。

如果我不停止脚本并且流保持打开状态，这个“问题”似乎不会发生，我只是没有在任何地方的文档中看到这一点，而且我已经看到人们将数据放入流的示例无需运行长期脚本。例如。提交表格后。

我刚开始使用 Kinesis，其 API 可在此处获得

我用它把100 条记录推送到kinesis

现在我想获取推送的记录数。为此，我正在使用这个：

现在使用上面的迭代器，我得到的计数是：

这是我的函数getDataCount(shardId, myStreamName)

但是这段代码每次运行时都会给出不匹配的结果，比如有时它会显示 81 有时 91

请对此有所了解.. :)

我知道 Kinesis Stream 中的数据记录将在 24 小时内自动删除。但是在我的应用程序中，当我将一些数据写入流时，如果我想第二次写入一些其他数据，首先插入的数据应该被删除。请任何人帮助我，因为我是使用 AWS Kinesis Stream 的新手...我没有从 Kinesis Service API 获得任何帮助...

好的，我将从一个详细的用例开始，并解释我的问题：

1. 我使用第 3 方 Web 分析平台，该平台利用AWS Kinesis流将数据从客户端传递到最终目的地 - Kinesis 流；
2. 网络分析平台使用 2 个流：
   1. 一个数据收集器流（单个分片流）；
   2. 第二个流，用于丰富来自收集器流（单个分片流）的原始数据；最重要的是，这个流使用TRIM_HORIZON迭代器类型消耗来自第一个流的原始数据；
3. 我使用AWS Java SDK使用流中的数据，具体使用GetShardIteratorRequest该类；
4. 我目前正在开发提取类，所以这是同步完成的，这意味着我只在编译类时才使用数据；
5. 该类令人惊讶地工作，尽管有些事情我无法理解，特别是关于如何从流中消耗数据以及每种迭代器类型的含义；

我的问题是我检索到的数据不一致，并且其中没有时间逻辑。

• 当我使用AT_SEQUENCE_NUMBER并提供分片中的第一个序列号时

  .getSequenceNumberRange().getStartingSequenceNumber();

  ...作为``，我没有得到所有记录。同样，AFTER_SEQUENCE_NUMBER;

• 当我使用LATEST时，我得到零结果；
• 当我使用TRIM_HORIZON应该有意义的使用时，它似乎无法正常工作。它曾经为我提供数据，然后我添加了新的“事件”（记录到最终流中）并且我收到了零记录。神秘。

我的问题是：

1. 如何安全地使用流中的数据，而不必担心丢失记录？
2. 有没有替代品ShardIteratorRequest？
3. 如果有，我如何才能“浏览”流并查看其中的内容以进行调试参考？
4. TRIM_HORIZON该方法缺少什么？

在此先感谢，我真的很想从 Kinesis 流中了解更多有关数据消耗的信息。

使用 Amazon Kinesis 连接器库，是否可以配置两个发射器？例如，我想从流中过滤和转换记录，然后使用 Redshift 发射器将结果存储在 Redshift 中，同时还将过滤和转换的记录发送到新的 Kinesis 流。

IKinesisConnectorPipeline 接口似乎只为单个发射器设计，但我会假设许多用户正在使用单个流执行多项操作，所以我想知道执行此操作的标准方法是什么。

我正在关注有关从 Amazon Kinesis 流式传输数据到 Amazon EMR 上的 Hive 的教程。我可以使用 Hive 内部表成功地做到这一点，但是当我尝试使用外部表时，我遇到了一个奇怪的 EOF 错误。我以前没有同时使用 STORED BY 和 STORED AS，我认为这可能是问题所在。我使用 IBM 指南（http://www-01.ibm.com/support/knowledgecenter/SSPT3X_2.1.1/com.ibm.swg.im.infosphere.biginsights.bigsql.doc/doc/bsql_create_table.html）来验证它们可以一起使用。

这是我用来构建 hive 外部表的代码。

这是错误：

我在 Amazon WebServices 中设置了 Kinesis 流。我还想完成以下任务：

1. 使用单个分片 (C# Api) 将记录放入单个流中 - 成功
2. 我还编写了示例应用程序，其中多个生产者正在处理不同的流 - 成功
3. 我还设置了示例应用程序来执行多个工作人员将数据放入单个流中 - 成功

我还希望能够在 Reacords 中强制执行 SequenceNumberOrdering。

但真正的痛苦是使用 Kinesis C# Api 的 GetRecords 消费者操作。

我为记录创建了一个示例应用程序。问题是即使 Kinesis Stream 中没有记录，它也不会停止迭代。此外，将 SequenceNumber 保存在数据库或某个文件中并再次检索文件非常耗时 - 将 Kinesis Stream 用于 GetRecords 有什么优势？

为什么即使 Stream 中没有数据，它也会继续迭代？

我使用以下代码作为参考；

我们正在使用 Amazon Kinesis（一种队列服务）并拥有用 Java 编写的队列读取器。他们基本上从队列中读取数据并将数据插入我们的数据存储区。我想知道是否有人成功地使用 New Relic 来监控后台队列工作人员？

我感兴趣的一些分析：

1. 现在有多少队列工作人员正在运行？（它们根据负载放大和缩小）
2. 每个队列工作人员每秒处理多少条消息？随着时间的推移，这看起来如何？
3. 整个工人队列每秒处理多少条消息？
4. 工作人员向 MySQL 和 Cassandra 发出请求。他们花在这方面的时间有多少？
5. 我们正在使用 log4j 进行日志记录。如果工人产生了错误/痕迹，它们是什么？随着时间的推移，错误率是多少？

谢谢，

高级

Is there a way to tell what a shard's status is, whether it's OPEN, CLOSED, or EXPIRED? The only way I've been able to determine this information seems to be attempting an operation on the shard.