时间戳

首页 > 解决方案 > 从数据库加载大量记录时内存不足

scala - 从数据库加载大量记录时内存不足

问题描述

我在 Akka Streams 中使用 slick 从数据库（postgresql）加载大量记录（~2M）并将它们写入 S3 文件。但是，我注意到我下面的代码适用于大约 50k 的记录，但对于超过 100k 标记的任何记录都失败了。

  val allResults: Future[Seq[MyEntityImpl]] =
    MyRepository.getAllRecordss()

  val results: Future[MultipartUploadResult] = Source
    .fromFuture(allResults)
    .map(seek => seek.toList)
    .mapConcat(identity)
    .map(myEntity => myEntity.toPSV + "\n")
    .map(s => ByteString(s))
    .runWith(s3Sink)

以下是myEntity外观示例：

case class MyEntityImpl(partOne: MyPartOne, partTwo: MyPartTwo) {
  def toPSV: String = myPartOne.toPSV + myPartTwo.toPSV
}
case class MyPartOne(field1: String, field2: String) {
  def toPSV: String = {s"$field1|"+s"$field2"}
}
case class MyPartOne(field1: String, field2: String) {
  def toPSV: String = {s"$field1|"+s"$field2"}
}

我正在寻找一种以更具反应性的方式执行此操作的方法，以免内存不足。

标签： scalaakkaslickakka-streamreactive-streams

解决方案

根本问题

问题是在将它们分派到s3Sink.

数据被拉入内存的第一个位置可能是您的MyRepository.getAllRecords()方法。大多数（如果不是全部）Seq实现都是基于内存的。您肯定使用本地内存的第二个地方是seek.toList因为 aList将所有数据存储在内存中。

解决方案

而不是Seq从getAllRecords 你那里返回 a 应该直接返回一个基于光滑的Sourceakka。这将确保您的物化流在进入 s3 之前只需要用于瞬态处理步骤的内存。

如果您的方法定义更改为：

def getAllRecords() : Source[MyEntityImpl, _]

然后流的其余部分将以反应方式运行：

MyRepository
  .getAllRecords()
  .map(myEntity => myEntity.toPSV + "\n")
  .map(ByteString.apply)
  .runWith(s3Sink)

推荐阅读