如何加快检查大量 ffdf 的重复项答案

【问题标题】：how to speed up checking duplication for huge ffdf如何加快检查大量 ffdf 的重复项
【发布时间】：2021-12-31 01:44:45
【问题描述】：

我有一个ffdf 的列表，如果将其加载到 RAM 而不是使用 ff 包，它会占用大约 76GB 的 RAM。以下是他们各自的dim()

> ffdfs |> sapply(dim)
         [,1]     [,2]     [,3]      [,4]      [,5]      [,6]      [,7]
[1,] 11478746 12854627 10398332 404567958 490530023 540375993 913792256
[2,]        3        3        3         3         3         3         3
         [,8]     [,9]     [,10]     [,11]    [,12]     [,13]     [,14]
[1,] 15296863 11588739 547337574 306972654 11544523 255644408 556900805
[2,]        3        3         3         3        3         3         3
        [,15]     [,16]    [,17]
[1,] 13409223 900436690 15184264
[2,]        3         3        3

我想检查每个ffdf中的重复次数，所以我做了以下操作：

check_duplication <- sample_cols |> sapply(function(df) {
    df[c("chr","pos")] |> duplicated() |> sum()
})

它可以工作，但速度非常慢。

我在 HPC 上，我有大约 110GB RAM 和 18CPU。

我是否可以调整任何其他选项或设置来加快流程？谢谢。

【问题讨论】：

你读过vignette("parallel", "parallel")吗？您可以相对轻松地并行化 sapply 调用。
给定 110GB 的 RAM，您真的可以将所有内容加载到内存中吗？那为什么不把所有的东西都转换成data.tables 然后做那些计算呢？在我的笔记本电脑上使用较小的数据集（1e6 行），转换后的计算速度大约快 30 倍。
@ekoam 我实际上试图将所有内容加载到 RAM 中，但是当我尝试计算 check_duplication 时，该过程花费了这么长的时间，整个 R 控制台通过 Linux 系统获取了 Killed。 . 我不知道为什么，但我怀疑是内存问题，所以我切换到ff，至少在几个小时后返回结果。
@MikaelJagan 我确实尝试过parallel::mclapply，但它给我带来了另一个问题：它不会返回每个元素结果。错误是Warning message: In mccollect(jobs) : 4 parallel jobs did not deliver results
看看my benchmark - sapply 比替代品更占用内存。扩展性不如替代品。

标签： r hpc ff ffbase

【解决方案1】：

并行化是加快这一进程的一种自然方式。可以通过data.table在C级别完成：

library("data.table")

data.table 1.14.2 using 4 threads (see ?getDTthreads).  Latest news: r-datatable.com

set.seed(1L)
x <- as.data.frame(replicate(2L, sample.int(100L, size = 1e+06L, replace = TRUE), simplify = FALSE))
y <- as.data.table(x)
microbenchmark::microbenchmark(duplicated(x), duplicated(y), times = 1000L)

Unit: milliseconds
          expr       min         lq       mean     median         uq       max neval
 duplicated(x) 449.27693 596.242890 622.160423 625.610267 644.682319 734.39741  1000
 duplicated(y)   5.75722   6.347518   7.413925   6.874593   7.407695  58.12131  1000

这里的基准测试表明，duplicated 在应用于data.table 而不是等效数据帧时要快得多。当然，快多少取决于您为data.table 提供的CPU 数量（请参阅?setDTthreads）。

如果您走data.table 路线，那么您将像这样处理您的 17 个数据帧：

nduped <- function(ffd) {
  x <- as.data.frame(ffd[c("chr", "pos")])
  setDT(x)
  n <- sum(duplicated(x))
  rm(x)
  gc(FALSE)
  n
}
vapply(list_of_ffd, nduped, 0L)

在这里，我们使用setDT 而不是as.data.table 来执行从数据帧到data.table 的就地强制转换，并且我们使用rm 和gc 来释放@987654337 占用的内存@ 在将另一个数据帧读入内存之前。

如果出于某种原因，data.table 不是一个选项，那么您可以坚持对数据帧使用duplicated 方法，即duplicated.data.frame。它没有在 C 级别并行化，因此您需要在 R 级别进行并行化，例如使用mclapply 将您的 17 个数据帧分配给批次并并行处理这些批次：

nduped <- function(ffd) {
  x <- as.data.frame(ffd[c("chr", "pos")])
  n <- sum(duplicated(x))
  rm(x)
  gc(FALSE)
  n
}
unlist(parallel::mclapply(list_of_ffd, nduped, ...))

此选项速度较慢，并且消耗的内存比您预期的要多。幸运的是，还有优化的空间。该答案的其余部分重点介绍了一些主要问题以及解决这些问题的方法。如果您已经选择了data.table，请随时停止阅读。

由于您有 18 个 CPU，您可以尝试同时处理所有 17 个数据帧，但由于一次将所有 17 个数据帧读入内存，您可能会遇到内存不足的问题。增加批量大小（即，将 17 个作业分布在少于 17 个 CPU 上）应该会有所帮助。
由于您的 17 个数据帧的长度（行数）差异很大，因此将它们随机分配到大小大致相同的批次可能不是一个好策略。您可以通过将较短的数据帧批处理在一起并且不将较长的数据帧批处理在一起来减少整体运行时间。 mclapply 有一个 affinity.list 参数给你这个控制。理想情况下，每个批次都需要相同的处理时间。

每个作业使用的内存量实际上至少是存储数据帧x 所需的内存量的两倍，因为duplicated.data.frame 复制了它的参数：

x <- data.frame(chr = rep(1:2, times = 5L), pos = rep(1:2, each = 5L))
tracemem(x)

[1] "<0x14babad48>"

invisible(duplicated(x))

tracemem[0x14babad48 -> 0x14babc088]: as.list.data.frame as.list vapply duplicated.data.frame duplicated

复制发生在方法主体中的vapply 调用内部：

duplicated.data.frame

function (x, incomparables = FALSE, fromLast = FALSE, ...) 
{
    if (!isFALSE(incomparables)) 
        .NotYetUsed("incomparables != FALSE")
    if (length(x) != 1L) {
        if (any(i <- vapply(x, is.factor, NA))) 
            x[i] <- lapply(x[i], as.numeric)
        duplicated(do.call(Map, `names<-`(c(list, x), NULL)), 
            fromLast = fromLast)
    }
    else duplicated(x[[1L]], fromLast = fromLast, ...)
}
<bytecode: 0x15b44f0f0>
<environment: namespace:base>

vapply 调用是完全可以避免的：您应该已经知道 chr 和 pos 是否是因素。我建议为duplicated.data.frame 定义一个替代品，它只做你的用例所必需的。例如，如果您知道 chr 和 pos 不是因子，那么您可以指定

duped <- function(x) {
  duplicated.default(do.call(Map, `names<-`(c(list, x), NULL)))
}

并计算 sum(duped(x)) 而不是 sum(duplicated(x))。事实上，你可以通过将 list 替换为 c 来做得更好：

fastduped <- function(x) {
  duplicated.default(do.call(Map, `names<-`(c(c, x), NULL)))
}

在此处使用 c 会导致数据帧 x 的行作为原子向量而不是列表进行存储和比较。换句话说，fastduped(x) 正在做

duplicated.default(<length-'m' list of length-'n' atomic vectors>)

而duped(x) 正在做

duplicated.default(<length-'m' list of length-'n' lists of length-1 atomic vectors>)

m = nrow(x) 和 n = length(x)。后者速度较慢，消耗更多内存，?duplicated 中有警告说：

将它用于列表可能会很慢，尤其是当元素不是原子向量（参见“向量”）或仅在属性上有所不同时。最坏的情况是 O(n^2)。

计算 sum(fastduped(x)) 而不是 sum(duplicated(x)) 应该会增加您可以同时处理的数据帧的数量，而不会耗尽内存。 FWIW，这是一个比较duplicated、duped、fastduped 运行时间的基准（只字不提内存使用）：

set.seed(1L)
x <- as.data.frame(replicate(2L, sample.int(100L, size = 1e+06L, replace = TRUE), simplify = FALSE))
microbenchmark::microbenchmark(duplicated(x), duped(x), fastduped(x), times = 1000L)

Unit: milliseconds
          expr      min       lq     mean   median       uq      max neval
duplicated(x) 521.7263 598.9353 688.7286 628.8813 769.6100 1324.458  1000
      duped(x) 521.3863 598.7390 682.1298 627.1445 764.7331 1373.712  1000
  fastduped(x) 431.0359 528.6613 594.1534 553.7739 609.6241 1123.542  1000

【讨论】：