代码之家 › 专栏 › 技术社区 › igx

akka将文件行流式传输到actor路由器,并使用单个actor进行写入。如何处理背压

alpakka akka-stream akka scala

0

igx · 技术社区 · 5 年前

我想将一个文件从s3流式传输到actor,以便对其进行分析和充实,并将输出写入另一个文件。解析器的数量应该受到限制,例如 application.conf

akka{
    actor{
        deployment {
              HereClient/router1 {
                router = round-robin-pool
                nr-of-instances = 28
              }
        }
    }
}

code

val writerActor = actorSystem.actorOf(WriterActor.props())
val parser = actorSystem.actorOf(FromConfig.props(ParsingActor.props(writerActor)), "router1")

但是,写入文件的参与者应限制为1(单例)

我试着做一些

val reader: ParquetReader[GenericRecord] =  AvroParquetReader.builder[GenericRecord](file).withConf(conf).build()

  val source: Source[GenericRecord, NotUsed] = AvroParquetSource(reader)
source.map (record =>  record ! parser)

但我不确定背压处理是否正确。有什么建议吗?

0 回复 | 直到 5 年前

1

0

nivox 5 年前

事实上,你的解决方案是忽略背压。

在保持背压的同时让流与参与者交互的正确方法是使用 ask pattern support of akka-stream ( reference ).

根据我对你的例子的理解,你有两个单独的演员互动点:

向解析参与者发送记录(通过路由器)
将解析后的记录发送给singleton write actor

我会做类似于以下的事情:

val writerActor = actorSystem.actorOf(WriterActor.props())
val parserActor = actorSystem.actorOf(FromConfig.props(ParsingActor.props(writerActor)), "router1")

val reader: ParquetReader[GenericRecord] =  AvroParquetReader.builder[GenericRecord](file).withConf(conf).build()

val source: Source[GenericRecord, NotUsed] = AvroParquetSource(reader)
source.ask[ParsedRecord](28)(parserActor)
      .ask[WriteAck](writerActor)
      .runWith(Sink.ignore)

我们的想法是你把所有的 GenericRecord 元素到 parserActor 这将以一个 ParsedRecord 例如,我们指定并行度为28,因为这是您配置的实例数,但是,只要您使用的值高于实际的参与者实例数,那么任何参与者都不应遭受工作不足。

一旦 parseActor 回复解析结果(这里用 ParsedRecord )我们应用同样的模式与单身作家演员互动。请注意,这里我们没有指定并行性,因为我们只有一个实例,所以一次发送超过1条消息是没有意义的(实际上,由于异步边界的缓冲,这种情况无论如何都会发生,但这只是一种内置优化)。在这种情况下,我们希望编剧和演员用 WriteAck 通知我们写作已经成功,我们可以发送下一个元素。

使用这种方法,你可以在整个流程中保持背压。

2

0

Jev Prentice 5 年前

我认为您应该使用一种“异步”操作

也许这另一个q/a能给你一些操作上的帮助 Processing an akka stream asynchronously and writing to a file sink