问题标签 [alpakka]

我有一个用 Scala/Akka 编写的应用程序并尝试添加 SOAP 支持。有人用 Alpakka 做过吗？或者最好的方法是什么。到目前为止，我认为骆驼是最好的解决方案。

我正在使用 Consumer.commitableSource 在 alpakka 中进行数据复制。但是，kafka 日志文件的大小增加得非常快。一天的大小达到 5 GB。作为这个问题的解决方案，我想立即删除处理过的数据。我在 AdminClient 中使用删除记录方法来删除偏移量。但是当我查看日志文件时，与该偏移量对应的数据并没有被删除。

我正在使用 Alpakka 将单个文件上传到 SFTP 服务器，但是一旦文件上传并且我得到了成功响应，Sink 保持打开状态，我该如何排空它？

我从这个开始：

但这最终永远不会离开地图步骤。我试图添加一个killswitch，但这似乎没有效果（无论是关闭还是中止）：

我在做一些根本错误的事情吗？我只想要一个结果。

编辑发送到的设置toPath：

我创建了一个 api 来向我发送文件，我会将它们上传到 S3，并且我使用akka-stream-alpakka-s3库来使用流来完成它。

我的问题是在我的控制器中我可以将文件转换为 Jave 文件：

在我的 s3 服务函数中，我只能使用 scala 文件，因为它有一个“toByteArray”函数，我需要它作为源代码，它看起来像这样：

对齐文件类型的最佳方法是什么，以便我能够将字节字符串文件传递给我的源？

我正在使用 Alpakka 解析 csv 文件。版本 "com.lightbend.akka" %% "akka-stream-alpakka-csv" % 0.20 我有带有未封闭报价的 csv 文件。

我想跳过坏行然后继续，但我的流量正在下降。

我正在使用 supervisorStrategy Supervision.Resume，但它不起作用。

找到未关闭的报价时流失败。

有没有办法解决这个问题？

我的代码：

我会直接潜入。我正在使用ScanamoAlpakka.

我有以下内容：

目标是迭代地图并删除任何空值。删除空值后，我需要将每个键/值对转换为基于其类型的查询条件。

像这样：

这不起作用，因为上面概述的条件不会立即成为查询条件类的对象。我一直在查看 SDK，并且很难找到一个可以将键/值对转换为可查询条件的类。

我尝试过使用Condition(key, value)which 似乎可以工作，但是当以这种格式完成时，似乎所有申请的隐式DynamoFormat都丢失了，因为我收到以下错误：

Error:(22, 45) could not find implicit value for evidence parameter of type com.gu.scanamo.query.ConditionExpression[(Symbol, Any)] case (key, Some(value)) => Condition(key -> value)

我们正在构建一个高吞吐量低延迟的流处理应用程序。我们将 Apache Kafka 用作消息传递平台和数据库。

Kafka Streams 和 Alpakka Kafka 框架似乎有很多共同之处，但 Kafka Streams 似乎对 Kafka 更“原生”，而 Alpakka 允许我们使用 Akka 框架的强大功能。

这两个框架之间的主要区别是什么？

我们使用 akka 流和 alpakka kafka 来使用来自不同主题的消息，使用主题模式订阅。使用 groupBy() 为每个主题创建子源/流。
我们如何暂停/恢复/控制单个主题的消息处理，而不影响其他主题的消息处理。

以下选项不起作用。
使用 RestartSource 将重新启动源并影响所有主题的处理。
使用 RestartFlow 会丢弃失败的元素，也会影响所有主题的处理。
如果我们必须为所有主题重新启动源并且有额外的开销，那么为每个主题创建一个源是资源密集型的。

我在使用 alpakka 的 XmlParsing Flow 时遇到问题

运行此程序时，图表会打印由 发出的元素contentFlow，这些元素是正确的并且与第一个文件的预期值匹配。之后，抛出此异常：

我了解这里发生的基本情况 - 解析器抱怨<?xml version="1.0" encoding="UTF-8"?>第二个文件顶部的指令的 ByteString，但我对流不够熟悉，不知道该怎么做。如果我删除该指令，我会得到一个稍微不同的例外，即拥有 2 个根元素。

我的目标是构建一个从某个位置读取文件并发出CustomContent以供进一步处理的图。我怎样才能修改它以将每个文件视为一个不同的输入单元？

我是 alpakka 的新手。我在 Alpakka 中使用以下代码使用 MongoDB 连接器来获取和循环 100K 记录

执行：

和路线：

我想在没有任何内存不足问题/堆空间的情况下将 mongodb 集合流式传输到 akka-http。请提出更好的方法。