从数据库加载大量记录时内存不足答案

【问题标题】：Running out of memory when loading large amount of records from database从数据库加载大量记录时内存不足
【发布时间】：2019-07-12 22:34:00
【问题描述】：

我在 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"}
}

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

【问题讨论】：

标签： scala akka slick akka-stream reactive-streams

【解决方案1】：

根本问题

问题是在将它们分派到s3Sink 之前，您要将数据库中的所有记录提取到本地内存中。

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

解决方案

而不是从getAllRecords you should be returning a slick-based akka Source directly 返回Seq。这将确保您的物化流在进入 s3 之前只需要用于瞬态处理步骤的内存。

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

def getAllRecords() : Source[MyEntityImpl, _]

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

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

【讨论】：