R - 按 block 读取 csv 并行处理它们

Question

如何按 block 读取 CSV 文件(它的大小太大而无法一次读取)，并使用 parallel 包处理所有 block ？假设我想按 block 计算一列的平均值。

如果没有 parallel，我会使用这样的东西:

library(readr)

f <- function(df_chunk, pos) mean(df_chunk$'mpg')

res <- read_csv_chunked(readr_example("mtcars.csv"), DataFrameCallback$new(f), chunk_size=10)

结果是:

> res
      [,1]
[1,] 20.37
[2,] 19.89
[3,] 20.39
[4,] 18.20

所以我希望这 4 个平均值由子进程计算。一些实现这一点的代码可能是这样的:

library(foreach)
library(doParallel)

registerDoParallel(6)

calc_avg <- function (iterable_df_chunks) {
  foreach(df_chunk = iterable_df_chunks, .combine = c) %dopar%
    mean(df_chunk$'mpg')
}

calc_avg(< some code with read_csv_chunked() >)

stopImplicitCluster()

谢谢!

Answer 1

好的，我找到了可行的解决方案。函数load_CPU只做一些 CPU 密集型计算来检查子进程是否真的完成了工作:

load_CPU <- function(n){
  i=3
  v=c(2)
  while (length(v)<=n-1){

    if (all((i%%v[v<ceiling(sqrt(i))])!=0)){ 
      v=c(v,i)
    }
    i=i+2;
  }
  return(v)
}

calc_avg <- function (path) foreach(y = read_csv_chunked(path, ListCallback$new(function (x, pos) x),
    chunk_size = 10), .combine = rbind, .export=c('load_CPU')) %dopar% {
        load_CPU(10000)
        mean(y$'mpg')
    }

calc_avg(readr_example("mtcars.csv"))

结果是:

          [,1]
result.1 20.37
result.2 19.89
result.3 20.39
result.4 18.20