处理 600 万个项目的列表

Question

我有一个包含 6+ 百万条记录的 CSV 文件。我正在读取 CSV 文件并收集数组列表中的每一行（列表）。整个列表需要经过一个分组逻辑，最后将每个分组推送到一个消息队列系统中。

这是我的代码：

var header: Map<String, Int> by Delegates.notNull()
val rows = csv.bufferedReader().useLines { lines ->
  lines
    .filter {
      validate(it)
    }
    .filterNotNull()
    .map { line ->
      convert(line)
    }
    .toList()
}
parse(rows, header)
  .forEach { 
    this.send(it)
  }

fun parse(rows: List<List<String>>, header: Map<String, Int>): List<Domain> {
return rows
  .asSequence()
  .map { row ->
    mapToColumn(row)
  }
  .mapNotNull { it.getOrNull() }
  .filter(this::someFilter)
  .groupBy { it['xyz'] }
  .map { groupedRow ->
    mapToDomain(groupedRow)
  }
  .mapNotNull { it.getOrNull() }
  .flatten()
  .toList()
}

使用 4GB RAM 和 1 个核心服务器，这会崩溃（耗尽服务器内存，我的应用程序会重新启动）。

这适用于较小的列表（200 万）条记录，但对于 6M 则失败。

我需要帮助来提高性能，也许是处理分组逻辑的更好方法，以减少内存消耗或使用不同类型的变量？

Answer 1

您可以用内存换取性能。您可能会多次读取 CSV 文件，每次通过获取一组元素（希望您有足够的内存来存储最大组的所有元素）：

val delimiter = "|"
val groupColumnIndex = 42

//read CSV line by line to capture only groups
val groups = csv.bufferedReader().useLines { lines ->
    lines
        .mapNotNull { it.split(delimiter).getOrNull(groupColumnIndex) }
        .toSet()
}

groups.forEach { group ->
    //read CSV line by line once again for each group
    val domainObjectsOfGroup = csv.bufferedReader().useLines { lines ->
        lines
            .mapNotNull { it.split(delimiter) }
            .filter { it.getOrNull(groupColumnIndex) == group }
            .filter { /* other filtering logic */ }
            .map { /* mapping to domain object*/ }
            .toList()
    }

    //send to MQS
    send(domainObjectsOfGroup)
}

Answer 2

我认为您使用序列的想法是正确的，以避免创建过多的中间列表 - 如果您的 RAM 用完，这可能是您的瓶颈。但是您并没有始终如一地使用它们，而是在几个地方创建列表和其他集合，这会占用内存。

这里有一些您至少可以查看的内容 - 我建议您查看某种基准测试，以便您可以跟踪您正在做出多少改进，即使它只是记录您处理了多少项目堆有多大！

---

  lines
    ...
    .toList()

首先，当您读入文件时，useLines 将它们作为序列 (lines) 传递给 lambda - 但您在最后调用 toList() 并创建一个包含所有 600 万行的列表文件。但是，您使用该列表 (rows) 做的第一件事就是再次从中生成一个序列！尝试删除toList() 并将rows 保留为Sequence

---

rows
  ...
  .groupBy { it['xyz'] }

此分组函数产生Lists 中的Map，包含序列中的每个项目。在此之前你已经进行了一些过滤，所以也许你没有剩下 600 万个，但除非你真的把你的数据集精简了，否则你将在这里有一张大地图——所有剩余的项目，并且取决于最终有多少组，可能还有数百万个列表。

根据您正在执行的具体操作，您或许可以避免一次做太多事情，从而导致大量结果占用您的内存。看看chunked 函数 - 它会生成Lists 的Sequence，因此您可以分段处理数据，并且在开始处理时会生成每个列表，而不是一次全部生成。

使用较小的收集可能会减轻一些内存压力，因为系统有更多机会进行垃圾收集。您可能可以按部分进行分组，进行域映射或其他操作，然后存储最终结果/将其传递给send 函数。或者调用toSequence，这样你的分块Sequence<List<Thing1>>最终变成Sequence<Sequence<Thing2>>，你可以flatten它。

我希望您能大致了解，这有点棘手，这实际上取决于您在做什么。这个想法基本上是尽可能地保持它的序列，对于需要创建集合的地方，你正在做“小”的，然后将它们变成一个序列，这样你就不会在内存中将它们全部收集在一起.

---

  .map { groupedRow ->
    mapToDomain(groupedRow)
  }
  .mapNotNull { it.getOrNull() }

你可以把这两个组合成

mapNotNull { groupedRow -> mapToDomain(groupedRow).getOrNull() }

您可以在阅读器中对链接的filter -> filterNotNull 调用执行相同的操作，但这里有一个Sequence，所以这并不重要，这只是一个额外的步骤。但是，当您使用Iterable 时，每个map 调用都会产生一个全新的集合，因此如果您可以使用一个而不是两个，它们会节省一些钱！我不知道这是否会简化为编译器在后台调用一次，但无论如何，尽你所能做你自己是值得的。

---

rows
  ...
  .flatten()
  .toList()

最后，您正在创建一个列表。即使您可以使用我提到的分块序列方法，最终结果是您将所有剩余的东西都收集到一个集合中。这还会大吗？太大了？

（如果不是太大，请忽略这一点，因为通常Sequence 的效率低于Iterable - 您需要进行基准测试并进行调用！）

您实际上对这个结果所做的是调用 forEach ，然后将每个项目传递给您的 send 函数 - 同样，这很容易成为 Sequence - 你不需要那个中介List，你所做的只是迭代它并对每个项目做一些事情，一次一个，这就是序列的好处。

toList 调用可能是多余的，这取决于 flatten 在这里返回的内容。如果它已经是一个列表，再次调用它可能会创建一个全新的列表，我不确定编译器是否会跳过它

---

那是很多东西，我只能给你看的东西，我不能真正告诉你如何根据你的情况来写它。希望它能给你一些指导！

tl;dr 尝试将事物保持为一个序列，注意将其转换为可迭代对象的调用，并尝试避免对这些调用重复步骤（因为每个步骤都会创建一个新集合） .如果你必须创建一个可迭代对象，请尽快查看是否可以toSequence 以避免创建更多，并研究分块。

另外看看您是否可以随时将内容写入存储。通过一个漂亮的直接序列将它流式传输到一个文件会很酷，但是如果你最终得到一个大列表，也许写它以供以后进一步处理是正确的做法。只是想一想！

好吧，这很长