问题标签 [amazon-kcl]

我正在探索 AWS Kinesis 的数据处理要求，用基于流的方法替换旧的批处理 ETL 处理。

该项目的关键要求之一是能够在以下情况下重新处理数据

• 发现并修复了一个错误，并重新部署了应用程序。数据需要从一开始就重新处理。
• 添加了新功能，并且需要全部或部分重新处理历史记录。

Kafka 的场景在这里得到了很好的记录 - https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Streams+Data+%28Re%29Processing+Scenarios。

我在 Kinesis 中看到了基于时间戳的 ShardIterator，我认为可以使用 Kinesis API 构建类似 Kafka 的重置工具，但如果这样的东西已经存在，那就太好了。即使没有，向解决过类似问题的人学习也是件好事。

那么，是否有人知道 Kinesis 中可用于执行此操作的任何现有资源、模式和工具？

我正在运行Ubuntu 16.04& Java 8。kcl生成数千条INFO日志行。有谁知道如何仅启用ERROR和WARN记录？

*我也有同样的问题kpl。

我没有日志文件。

我们有多个消费者从同一个流中读取。所以每次有新的消费者，DynamoDB 中都会创建一个新表。这些消费者不是永久的。一旦它们被关闭，表就会变得多余。

1.如果我们在消费者之间保持应用程序名称相同，则无法正确读取数据记录。2. 如果我们为每个消费者使用不同的应用程序名称，DynamoDB 就会被表淹没。

有没有办法处理这种情况？

我经历了几个问题和 SO 问题，但无法让示例应用程序正常工作。

以下是如何重现这一点；

以下是我在 prod.properties 中更改的唯一内容

在sample_kclpy_app.py我的process_record函数中是这样的；

我的日志

当我./bin/kcl-bootstrap -p app.properties -j /usr/bin/java -e在终端运行 KCL 时，我做ps -ef | grep java了并得到了它的 pid。我发出kill -s TERM $PID_ABOVE. 我注意到节点 recordProcessor 设法像这样记录 2018-03-31T11:13:46.998 INFO recordProcessor - Shutdown requested 2018-03-31T11:13:48.024 INFO recordProcessor - Shutting down...

当我通过 docker run 命令运行等效命令时 ，docker run -v /tmp:/tmp -v ~/.aws:/root/.aws -ti cde3946e2cf9发出Ctrl-C或终止 docker 并不会导致上面生成日志。例如，recordProcessor 没有收到有关关闭请求的通知。docker stop cde3946e2cf9docker kill --signal=SIGTERM cde3946e2cf9

我们的 docker 部署在 Kubernetes 集群中，当我们重新部署 docker 时，我可以看到Process terminanted, will initiate shutdown.主处理器正在记录它，但没有报告工作人员关闭请求。

我想在重新部署 docker 时捕获这些事件，以确保我们能够优雅地处理关闭。

任何人都知道如何确保在集群中重新部署 docker 时工作人员确实得到通知？

谢谢

我有一个运动流，它正在用 putRecord 连续写入。在消费者中，我使用 KCL 的 processRecords 进行消费。只要流中有记录，就会调用此流程记录。我需要每 X 小时执行一次此流程记录。

这是我尝试过的事情。

这似乎不起作用。它抛出以下异常

有人可以让我知道是否有解决办法来实现这一目标吗？谢谢

我正在尝试控制在 KCL 的 processRecords 方法中获取的记录数，这是我的 KCL 配置

我的 config.getMaxRecords() 设置为 500，但我仍然在 processRecords 方法中获得约 1000-1500 条记录，我希望严格的上限为 500，因为我的下游进程无法处理更多。

我在正式文档中找不到与检查点之间的AWS Kinesis任何明确引用，TRIM_HORIZON以及与检查点之间的任何引用LATEST。

你能证实我的理论吗：

• TRIM_HORIZON- 如果应用程序名称是新的，那么我将读取流中可用的所有记录。否则，应用程序名称已被使用，那么我将从上一个检查点读取。

• LATEST- 如果应用程序名称是新的，那么我将读取订阅流后添加的流中的所有记录。否则，应用程序名称已被使用，我将读取上一个检查点的消息。

• TRIM_HORIZON和之间的区别LATEST仅在应用程序名称是新的情况下。

我想了解处理 kinesis 消费者中断的最佳方法是什么。

我的消费者设置了 TRIM_HORIZON。以下是来自 AWS 的定义：

TRIM_HORIZON - 从最后一个（未修剪的）流记录开始读取，这是分片中最旧的记录。在 DynamoDB Streams 中，数据保留期限为 24 小时。年龄超过此限制的流记录将从流中删除（修剪）。

我对这个定义的理解是，如果我的消费者关闭然后重新启动。它将从分片中最旧的可读记录开始消费。这意味着它将重新读取一些旧数据。

我希望我的消费者在最后一个确认记录之后开始。

我的 kinesis 消费者使用 KCL (java) 并由 dynamoDB 支持。

我应该使用 AFTER_SEQUENCE_NUMBER 并在中断后给我的消费者一个序列号吗？如何获得最后一次阅读记录？我的目标是从消费者停机前的最后一个确认记录恢复消费记录？

我很欣赏你的建议。

谢谢，

穆莱。

我有一个用 KCL 用 Ja​​va 编写的 Kinesis 消费者。从消费者方面，我可以使用 setMaxRecords 方法限制速率，以设置一批中记录的最大限制。

我的问题是：如果我不设置任何限制，那么确定一批中有多少条记录的逻辑是什么？允许的最大值是否有默认值？