问题标签 [amazon-kinesis-firehose]

我正在使用包含客户交易的 AWS Kinesis Stream。我想在开始处理之前将事务备份到 S3 中。如何在不运行任何 Lambda 或其他计算组件的情况下使用裸 Kinesis Firehose 从 Kinesis Stream 备份事务？

我们通过 AWS Lambda-->Kinesis Firehose-->Redshift 摄取大量事件数据。我的问题; 无论如何，是否可以按时间（或其他检查点）从 s3 重新摄取 Firehose 流数据，比如我们是否想将流数据加载到另一个集群？

对于从我的 AWS Kinesis Firehose 到 AWS Kinesis Analytics 的数据流，我如何创建一个实时图形代码，每 2-3 秒更新一次，就像 NASDAQ 一样？如何从 Kinesis Analytics 获取输出？试图实现接近此的目标。

看起来微不足道，但我无法让它工作：我正在将编码为 JSON 对象的数据发送到调用 Firehose 的 API Gateway。S3 中生成的文件在一行中包含所有 JSON 对象。我读过应该可以添加换行符，但是无论我尝试什么，要么出现错误，要么根本没有做任何事情。映射模板如下所示

有什么想法该怎么做？

我正在为 Kinesis Firehose 使用 Java API，并尽可能利用 putRecordBatch()。但是，在我的应用程序中，我偶尔只发送一条记录，无法等待更多或将其缓存在内存中。

但是，我的代码比我想要的要复杂一些，因为我正在跟踪每个操作的成功/失败计数。putRecordBatch 返回一个 int 值，表示在发送的记录集中提交失败的次数，例如 4 次失败中的 2 次，而 putRecord() 要么成功，要么抛出异常。

我宁愿在任何地方都使用 putRecordBatch() 但文档说只有在您有多个记录要提交时才使用它。

这可能有点远，但是这里有没有人有这个 API 的经验，特别是这些方法？您是否已经完成了我正在考虑的事情并且只是在任何地方都使用了 putRecordBatch()，即使记录集为 1？

今天，当我通过 Kinesis、Firehose 和 Redshift 完成第一个学习步骤时，我很高兴地发现 Amazon 有一个“试用我们的演示数据生成器”设置。

我很沮丧地得知它似乎并没有真正起作用。

于是，我去挖了。而且我发现STL_LOAD_ERROR包含错误，表明需要一个分隔符，并且记录的前面看起来像{field:val,field:val}{field:val,field:val}.

“一定是在某处剥离换行符，”我想。

经过挖掘，我发现在相关的 S3 存储桶中有生产记录，格式令人惊讶：

也就是说，明显的记录之间没有分隔符，它们是一个几十K的单行文件。

其他 SO 帖子似乎表明这实际上是预期的数据格式。

为什么 Redshift 需要数据演示不使用的格式的数据？我要重新配置哪个？

我的用例是编写一个通用的 CSV 转换器，它应该能够将任何 Java POJO 转换为 CSV 字符串。

我的实施：

我正在使用 Jackson-dataformat-csv 库，但我坚持使用！HOW_TO！部分，即如何从objectList中提取对象的.class。我正在学习并遇到类型擦除，所以我认为除了将 .class 作为我的函数的参数之外，这是不可能的。但我也使用 Java 反射从通用实体中提取此对象列表，因此我无法选择提供 .class 参数。

有解决方法吗？

或者

List<T> objectList to List<String> csvList我可以转换具有添加分隔符、引号字符、行分隔符等功能的泛型的任何其他方法/库。

谢谢！

我有 Kinesis Firehose 流将数据加载到 Elasticsearch 集群中。该数据具有纬度和经度字段。在 Elasticsearch 中，我想在地图中可视化这些数据。但是 Elasticsearch 只接受 geo_point 字段类型。

如何将纬度和经度字段映射到 Elasticsearch 中的地理点字段？

谢谢

• AWS CLI API 将Kinesis Firehose的RedshiftDestinationConfiguration定义为具有RetryOptions参数。 参考

• cloudformation 架构还定义了Kinesis Firehose的RedshiftDestinationConfiguration，但没有RetryOptions参数。 参考

有没有办法设置和更新通过 cloudformation 文件维护的 firehose 的重试选项？

我的要求是从 AWS 收集 VPC 流日志。我创建了一个 VPC 流日志并将其关联到云监视日志组“流日志组”。从 CloudWatch 日志订阅中，我想通过 AWS Kinesis firehose 订阅它，但在订阅列表中我找不到“Kinesis Firehose”，只有“Lambda”和“ElasticSearch Service”。你能帮我解决这个问题吗？

我们不能使用 AWS Kinesis firehose 订阅 CloudWatch 日志。