Я использую slick в Akka Streams для загрузки большого количества записей (~ 2M) из базы данных (postgresql) и записи их в файл S3. Тем не менее, я заметил, что мой код ниже работает для записей около ~ 50 тыс., Но не работает для чего-либо выше отметки 100 тыс.
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"}
}
Я ищу способ сделать это более реактивным способом, чтобы он не исчерпал память.