问题标签 [amazon-kinesis-firehose]

我正在考虑使用 Firehose 从 Web 应用程序发送数据。

Firehose 服务通过 AWS 开发工具包公开

SDK 很重（aws-sdk-2.5.5.min.js 重 1 个月）。这可能会增加网站的加载时间。

这是一个问题吗？如何解决？是否可以只加载 SDK 的一部分？

我正在使用 Amazon Kinesis 代理来跟踪我的 nginx Web 访问日志，以便将访问日志推送到 Kinesis - 最终在 Kinesis Analytics 中使用。

代理正在运行并跟踪我的访问日志，然后将记录推送到 Kinesis。我可以看到到达 Kinesis 的记录，因此连接良好。

我遇到的问题与格式化程序选项有关 - 特别是关于“ LOGTOJSON ”。

以下是 的内容/etc/aws-kinesis/agent.json：

这是我用来以 apache 组合日志格式输出日志的 nginx 配置部分：

以下是 nginx 访问日志的几行输出：

这是 Kinesis 代理的输出。您可以看到记录正在“解析”但没有“转换”。

Kinesis Analytics 将传入数据检测为 CSV 流而不是 JSON 流。当我在 AWS 控制台中查看原始流内容时，它看起来像这样：

所以看起来代理正在发送每条记录而没有任何类型的转换，并用逗号分隔每一行。

我看不出我在哪里出错了。nginx 访问日志看起来是以正确的格式编写的，但 Kinesis 代理不会将其转换为 JSON - 原因不明。

任何指针将不胜感激。

有没有办法使用 Lambda 进行 S3 文件连接？

我有 Firehose 以最长的可能间隔（15 分钟或 128mb）将数据流式传输到 S3，因此我每天有 96 个数据文件，但我想将所有数据聚合到一个每日数据文件中，以便在稍后读取数据时获得最快的性能在 Spark (EMR) 中。

我创建了一个解决方案，当 Firehose 将新文件流式传输到 S3 时，会调用 Lambda 函数。然后该函数从源存储桶中读取 (s3.GetObject) 新文件，并从目标存储桶中读取连接的每日数据文件（如果它已经与以前的每日数据一起存在，否则创建一个新文件），将两个响应主体解码为字符串，然后只需将它们加在一起并使用 s3.PutObject 写入目标存储桶（它会覆盖以前的聚合文件）。

问题是当聚合文件达到 150+ MB 时，Lambda 函数在读取两个文件时达到其 ~1500mb 内存限制，然后失败。

目前我的数据量很少，每天几百 MB-s，但这个数量在未来会呈指数增长。对我来说很奇怪 Lambda 的限制如此之低，并且已经用如此小的文件达到了它们。

或者连接 S3 数据的替代方案是什么，理想情况下由 S3 对象创建事件或以某种方式调用的计划作业，例如每天计划？

我遵循了几个教程并添加了不同的 IAM 配置和安全组，以使事情尽可能开放和明显，但没有任何效果。这是调试日志，主要错误粘贴在下面。让我知道我是否应该添加任何其他数据来明确这一点。

我们正在尝试将 AWS Kinesis Firehose 与 Redshift 输出一起使用。我们已经创建了 Firehose Delivery Stream。我们在 VPC 安全组中有一个 Redshift 集群。但似乎 Firehose 无法连接到集群。

这是我们创建 VPC 安全组的片段。

在这个链接之后，我们设置了一个允许来自“52.19.239.192/27”的连接的入口规则。但这似乎不起作用，我们仍然收到以下错误。

与指定 Amazon Redshift 集群的连接失败。确保安全设置允许 Firehose 连接，确保 Amazon Redshift 目标配置或 JDBC URL 中指定的集群或数据库正确，并且集群可用。

有谁知道我们做错了什么？

我只选择 AWS.firehose 服务来使用 SDK Builder for javascript（此处）。

当尝试从 node.js 使用它时，我运行了以下代码：

并得到这个错误：

/aws-sdk-2.6.6.min.js:7 if(this._events||(this._events={}),"error"===t&&(!this._events.error||s(this. _events.error)&&!this._events.error.length)){if(e=arguments 1 ,e instanceof Error)throw e;var l=new Error('Uncaught, unspecified "error" event. ('+e+" )");throw l.context=e,l}if(r=this._events[t],a(r))return!1;if(i(r))switch(arguments.length){case 1: r.call(this);break;case 2:r.call(this,arguments 1 );break;case 3:r.call(this,arguments 1 ,arguments 2 );break;default:o=Array.prototype。 slice.call(arguments,1),r.apply(this,o)}else if(s(r))for(o=Array.prototype.slice.call(arguments,1),c=r.slice() ,n=c.长度,u=0;u

ReferenceError：AWS 没有在 Object 中定义。(/aws-sdk-2.6.6.min.js:7:25500) 在 Module._compile (module.js:541:32) 在 Object.Module._extensions..js (module.js:550:10) 在Module.load (module.js:458:32) at tryModuleLoad (module.js:417:12) at Function.Module._load (module.js:409:3) at Module.require (module.js:468:17) ) 在 Object 的 require (internal/module.js:20:19) 处。(/index.js:5:11) 在 Module._compile (module.js:541:32)

这是来自 AWS SDK builder 的错误吗？

我应该包括其他服务吗？

还有比这更多的文档吗？

我目前正在使用以下映射模板将发送到 AWS API Gateway 端点的数据传递到 AWS Kinesis Firehose 流：

我想做的是：向$input.body正在编码的信息添加信息，就像$context.identity.sourceIp发出请求的客户端一样。

当传递给 Kinesis Firehose 的输出需要进行 Base64 编码时，我该如何处理？理想情况下，我希望发布到 Kinesis Firehose 的数据如下所示：

我正在使用 Dynamo DB 流 + Lamdba 作为触发器来调用 kinesis，它将我的 Dynamo DB 数据放入 Redshift。

有人可以建议一种使用发电机流将 Dynamo DB 数据加载到不同区域的 Redshift 的方法。

查看 amazon kinesis 分析的营销文献，我正在查看他们的实时日志分析流程，他们有第四步，将数据通过管道传输到 Amazon ES 以获取内容建议。

什么是亚马逊 ES？那是他们的弹性搜索服务吗？如果是这样，弹性搜索如何生成个性化推荐？

我们通过 API-Gateway > Kinesis Streams > Lambda > FireHose > Redshift Tables 使用所有分析源。

AWS Lambda 是我们的转换足迹，它接受 Kinesis 流记录根据上下文更改分析事件并将其拖放到 FireHouse 以将其保存到 Redshift DB。

在这个旅程中，我们希望在某些条件下更新 Redshift 记录（基本上是 UPSERT ( Insert | Update ) ）。firehose 中是否有任何东西可以避免 redshift 中的重复记录？