问题标签 [amazon-kinesis]

以这样的方式存储数据的最佳方式是什么，以便我可以获得查询的实时答案，例如“给我统计过去 2 周失败的交易”、“统计过去 2 年内创建的帐户”。每次计算行数不是一种选择，因为表中的单个条目数量很大，并且可能需要数小时来计算。

我只对以滚动窗口方式实时查找聚合感兴趣。此外，我不想保留超过 2 年的数据，并希望自动将其删除。

有没有解决这个问题的标准方法？redshift/kinesis 之类的服务有帮助吗？

感谢期待。

我有以下 Python 块来获取已放置到 Kinesis 流中的记录，然后将记录放入 S3 存储桶中。此流有一个分片。

数据最初是从流中提取并推送到 S3，但在某些时候，我TypeError: expected string or buffer在线接收response = kinesis_connection.get_records(shard_iterator)。分片上还有更多记录可供我提取。以前有没有人遇到过这个问题和/或是否有人对我应该尝试解决的问题有什么想法？

从很多来源我计划使用 Amazon kinesis 来捕获流，并且在一定级别的数据转换之后，我想在某些表模式中将流定向到 Redshift Cluster。在这里我不确定这样做是否正确？

从 Kineis 文档中，我发现它们具有与红移的直接连接器。但是我还发现，如果我们进行批量上传，Redshift 看起来会更好，因为数据仓库系统需要索引。所以建议将所有流存储到 S3，然后 COPY 命令在 redshift 上进行批量推送。有人可以添加更多视图吗？

我有 Kinesis Stream 捕获的事件。我想将所有事件放在 S3 上的特定文件夹结构中。我想制作一个带有日期戳的文件夹，就像 6 月 15 日的所有事件都应该进入该文件夹，而 6 月 16 日之后的新文件夹应该来选择事件等等。

作为 Kinesis 的新手，我只是使用文档，我发现有一个连接器框架，其中 S3Emitter 与配置一起使用来选择需要发出数据的 S3 位置。但是有人可以建议我如何维护文件夹结构在日期明智的文件夹中捕获事件日期？

我是 Kinesis 的新手。阅读我发现的文档，我可以创建 Kinesis Stream 以从 Producer 获取数据。然后使用 KCL 将从 Stream 中读取此数据以进行进一步处理。我了解如何通过实现 IRecordProcessor 来编写 KCL 应用程序。

然而，我仍然不清楚如何将数据放在 Kinesis 流上的第一阶段。我们是否有一些 AWS API 需要实施才能实现这一点。

场景：我有一台服务器，它不断地从文件夹中的各种来源获取数据。每个文件夹都包含文本文件，其行包含进一步分析工作所需的属性。我必须将所有这些数据推送到 Kinesis Stream。

我需要如下代码，类 putData 方法将用于在 Kinesis 流中输出

但是从服务器的源文件夹中读取文件，然后我应该使用什么设计来调用 putData 以获取 Kinesis 流上的记录。我是否需要无限循环并读取所有文件，然后执行此操作或某些框架会更好地做到这一点，同时兼顾容错性和单点故障。任何帮助将不胜感激。

简而言之：我需要一种更好的技术来将定期生成的数据放到 Kinesis Stream 中，这些数据会定期生成到服务器。谢谢

我有一个包含文件夹日期明智的服务器，每个文件夹还包含许多文件（每个大小 200kb），其中包含特定日期的所有日志。我是 RabbitMQ 的新手，在浏览 RabbitMQ 的文档时，我在下面找到了 Producer 的代码

参考链接：https ://github.com/rabbitmq/rabbitmq-tutorials/blob/master/java/Send.java

在上面的代码中，我添加了示例字符串“Hello World！” 发布。如上面问题描述中所述，我必须从具有不同日期戳目录的服务器读取日志信息所以我需要编写一个简单的无限循环（因为日志不断更新）并递归读取所有目录和文件然后对于文件的每一行，我可以撰写一条消息，然后将其发布给接收者？

在这种情况下，我们的通道将永远不会关闭并且 Connection 将始终处于运行状态，这是 RabbitMQ 的空闲状态吗？

RabbitMQ 是否可以标记已读取的文件并且不再读取它，或者我需要以编程方式管理它，例如用一些不同的名称重命名文件和文件夹。我在想这可能是我们的程序在我处于任何文件中间时因电源故障或其他原因而终止，然后我如何保证记录不会重复？

任何其他实现这一目标的最佳方法对我来说都会有很大帮助。提前致谢。

每个 Kinesis 应用程序都必须包含以下三个组件：

1. IRecordProcessor 接口

2. 实现 IRecordProcessor 接口的类的工厂

3. 初始化应用程序并创建工作者的代码

现在我在这里了解到，一旦我们将生产者配置为将记录添加到 Kinesis 流。然后 KCL 应用程序可以通过以下 processRecords 实现从 Kinesis 流中读取记录。那么这个 processRecords 方法应该有某种方法将它传递给 S3 以使用连接器库进行最终存储。

public void processRecords（列出记录，IRecordProcessorCheckpointer checkpointer）

查询：我应该如何从 KCL 应用程序的 processRecords 调用连接器库来将数据记录存储在 S3 上？

我浏览了显示示例 Kinesis 应用程序的链接 https://github.com/aws/aws-sdk-java/blob/master/src/samples/AmazonKinesisApplication/SampleRecordProcessor.java

在上面的链接中，我正在粘贴方法片段，如下所示。

在上面的代码中，当我们说“处理记录的逻辑在这里”。（请参阅上面的代码）这里我的要求是将数据放在 s3 上。我知道我们有可以执行此操作的连接器库，但是我现在无法想象如何调用连接器库？请建议

我正在研究 Amazon Kinesis Api 和 Kinesis 客户端库，我创建了一个生产者将数据放入流中，并有多个消费者应用程序从该流中读取数据。

我有根据输入流大小和输出流大小以及使用消费者应用程序计数动态增加和减少流大小的场景。

我找到了一些有用的来源来计算来自亚马逊网站的碎片数量，但不知道如何计算。源 URL 是： http ://docs.aws.amazon.com/kinesis/latest/dev/how-do-i-size-a-stream.html

需要对此有所了解。

谢谢

我试图了解如何部署使用 Kinesis 客户端库 (KCL) 构建的 Amazon Kinesis 客户端应用程序。

我发现了这个，但它只说明

在部署 Amazon Kinesis 应用程序时，您可以按照自己的最佳实践将代码部署到 Amazon EC2 实例。例如，您可以将您的 Amazon Kinesis 应用程序添加到您的 Amazon EC2 AMI 之一。

这并没有给我一个更广泛的画面。

这些示例使用 Ant 脚本来运行 Java 程序。这是最好的做法吗？

此外，我什至在运行 EC2 实例之前了解我需要确保

1. 开发的代码 JAR/WAR 或任何其他格式需要在 EC2 实例上
2. EC2 实例需要具备所有必需的环境（例如 Ant 设置）才能执行程序。

有人可以添加更多细节吗？

如何在本地使用 pig 脚本从 kinesis 流中提取数据。

我从 amazon 文档中注意到了这个示例，但不确定如何导入 amazon kinesis pig 库，并且该示例似乎不完整。我在哪里设置凭据，我在哪里可以获得 kinesis 库的 jar ......等等......他们的例子来自 grunt-shell，但我如何从我自己的机器上本地运行它？

http://docs.aws.amazon.com/ElasticMapReduce/latest/DeveloperGuide/kinesis-pig-generate-data.html