问题标签 [amazon-kinesis]

我已经用 python 设置了我的电脑并连接到 AWS。这已使用 s3_sample.py 文件成功测试，我必须使用文件中的凭证创建一个 IAM 用户帐户，该文件适用于 S3 存储桶。

我的下一个任务是创建一个 mqtt 桥，并使用 awslab - awslabs/mqtt-kinesis-bridge将一些数据放入 kinesis 流中。这似乎没问题，除了我在运行 bridge.py 时遇到错误。错误是：

奇怪的是，这不是我在建议为此桥设置的 .boto 文件中使用的帐户，这与我用于 S3 存储桶的凭据相同

在我看来，bridge.py 有一个硬编码的帐户，但我看不到它，也看不到它指向 .boto 文件的凭据。提前致谢

如何判断阅读器已处理 Kinesis 流中数据的百分比？我知道每个阅读器都有一个每个碎片检查点的序列号，我也可以从 describe-stream 中获取每个碎片的 StartingSequenceNumber，但是，我不知道阅读器当前在我的数据中有多远（我不知道知道分片的最新序列号）。

我正在考虑为每个分片获取一个 LATEST 迭代器并获取最后一条记录的序列号，但是如果自从我得到 LATEST 迭代器以来没有新数据，这似乎不起作用。

有什么想法或工具可以做到这一点吗？

谢谢！

我是 Amazon Kinesis 的新手。我已阅读基本文档和 Kinesis APi 以开始使用 Kinesis。所以我在 scala 中编写了一小段代码来创建一个流。但我面临如下异常

线程“主”java.lang.NoSuchMethodError 中的异常：org.joda.time.format.DateTimeFormatter.withZoneUTC()Lorg/joda/time/format/DateTimeFormatter；在 com.amazonaws.auth.AWS4Signer.(AWS4Signer.java:44) 在 sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) 在 sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:57) 在 sun.reflect.DelegatingConstructorAccessorImpl。 newInstance(DelegatingConstructorAccessorImpl.java:45) at java.lang.reflect.Constructor.newInstance(Constructor.java:526) at java.lang.Class.newInstance(Class.java:379) at com.amazonaws.auth.SignerFactory.createSigner (SignerFactory.java:119) 在 com.amazonaws.auth.SignerFactory.lookupAndCreateSigner(SignerFactory.java:105) 在 com.amazonaws.auth。

这是我的一段代码

这是我在 pom 文件中的依赖项

我正在使用带有 kinesis 的火花流，并在运行代码时遇到此异常

这是我的代码

并面临以下异常

我正在关注http://docs.aws.amazon.com/kinesis/latest/dev/kinesis-using-sdk-java-add-data-to-stream.html上的教程- 将 PutRecords 示例从 java 转换为斯卡拉。

我已经包含amazon-kinesis-client v1.20和aws-java-sdk-kinesis v1.9.16作为依赖项。

使用 sbt 编译会出现以下错误：

value putRecords is not a member of com.amazonaws.services.kinesis.AmazonKinesisClient

使用控制台进行试验确认，这些库随附的 AmazonKinesisClient 版本上似乎不存在此方法。有没有其他人遇到过这个问题，有人知道如何解决吗？

我正在尝试创建一个执行 Hive 查询的 AWS Datapipeline，其输出被写入 S3 存储桶。然后将数据从 AWS Redshift 集群中的 S3 存储桶中移出。

Hive 查询使用 Kinesis Stream 作为其输入。我正在尝试利用 Kinesis 的检查点功能，并希望将一个变量传递给我的 ShellCommandActivity，以便我可以在配置单元脚本中设置迭代次数。无论如何，我是否有一个每次运行数据管道时都会增加 1 的变量？

任何帮助都会很棒！

我正在检查有关事件中心中消息保留的详细信息。假设，我已将保留政策设置为 1 天，并且我已经发送了一些消息。那么，如果我将消息retentionPolicy 更改为3 天，现有的eventData也会保留3 天吗？

我们有 2 个生产者和 1 个消费者，我们使用 KCL 处理来自 kinesis 的数据滞后 24 小时。我们如何提高性能。

我正在尝试做的事情：最终：我想从浏览器扩展（Safari、Chrome）填充 AWS Kinesis 流。我需要使用签名过程 (v4) 将请求发送到 AWS；这涉及设置标头并加密它们（在具有 aws 密钥的远程服务器上）以最终将它们加入请求。

亚马逊要求明确定义标头“主机”......但是 Javascript 严格禁止设置它（以及其他一些，有充分的理由）

我一定是错过了什么——我该怎么做？

来源： http ://docs.aws.amazon.com/general/latest/gr/sigv4-signed-request-examples.html - 该示例使用 Python，但它显示了他们打算如何在服务器端使用 Host 标头

注意：我目前正在使用 POST；描述 GET 和查询字符串的示例注释也提到了“主机”：它必须作为请求中的标头存在

另外：使用类似的设置，我设法将文件从客户端直接上传到 S3 存储桶——自动化过程略有不同（不需要“主机”）。

我正在按照此处的 Scalazon 示例创建 Kinesis 流。下面的一段代码：

给出以下错误：

如果我不使用 for comprehension 和 call val createStream = Kinesis.streams.list，则没有错误。似乎无法弄清楚为什么。

同样，以下代码：

产生类似的错误：

感谢帮助！