【问题标题】:R - apply function on each element of array in parallelR - 对数组的每个元素并行应用函数
【发布时间】:2019-07-02 03:44:44
【问题描述】:

我有最高和最低温度以及降水的测量值,这些测量值按大小排列 (100x96x50769),其中 i 和 j 是关联坐标的网格单元,z 表示一段时间内的测量次数。

从概念上讲,它看起来像这样:

我正在使用climdex.pcic 包来计算极端天气事件的指数。给定最高和最低温度和降水的时间序列,climdexInput.raw 函数将返回一个climdexIput 对象,该对象可用于确定多个指标:霜冻天数、夏季天数、连续干旱天数等。

函数的调用非常简单:

ci <- climdexInput.raw(tmax=x, tmin=y, prec=z,
                       t, t, t, base.range=c(1961,1990))

其中 x 是最高温度的向量,y 是最低温度的向量,z 是降水的向量,t 是带有测量 x、y 和 z 的日期的向量。

我想做的是为我的数组的每个元素(即上图中的每个网格单元)提取时间序列,并用它来运行climdexInput.raw 函数。

由于真实数据的元素数量众多,我想在我的 4 核 Linux 服务器上并行运行这个任务。但是,我对 R 中的并行化没有经验。

这是我的程序的一个示例(特意减小了尺寸以在您的计算机上更快地执行):

library(climdex.pcic)

# Create some dates
t <- seq(as.Date('2000-01-01'), as.Date('2010-12-31'), 'day')

# Parse the dates into PCICt
t <- as.PCICt(strftime(t), cal='gregorian')

# Create some dummy weather data, with dimensions `# of lat`, `# of lon` and `# of timesteps`
nc.min <- array(runif(10*9*4018, min=0, max=15), c(10, 9, 4018))
nc.max <- array(runif(10*9*4018, min=25, max=40), c(10, 9, 4018))
nc.prc <- array(runif(10*9*4018, min=0, max=25), c(10, 9, 4018))

# Create "ci" object
ci <- climdexInput.raw(tmax=nc.max[1,1,], tmin=nc.min[1,1,], prec=nc.prc[1,1,],
                       t, t, t, base.range=c(2000,2005))

# Once you have “ci”, you can compute any of the indices provided by the climdex.pcic package.
# The example below is for cumulative # of dry days per year:
cdd <- climdex.cdd(ci, spells.can.span.years = TRUE) 

现在请注意,在上面的示例中,我仅使用数组的第一个元素 ([1,1,]) 作为 climdexInput.raw 函数中的示例。

如何利用并行处理对所有元素执行相同的操作,可能通过循环我的数组的维度ij

【问题讨论】:

    标签: arrays r parallel-processing multicore


    【解决方案1】:

    您可以使用 foreach 来做到这一点:

    library(doParallel)
    registerDoParallel(cl <- makeCluster(3))
    res <- foreach(j = seq_len(ncol(nc.min))) %:% 
      foreach(i = seq_len(nrow(nc.min))) %dopar% {
        ci <- climdex.pcic::climdexInput.raw(
          tmax=nc.max[i,j,], 
          tmin=nc.min[i,j,],
          prec=nc.prc[i,j,],
          t, t, t, 
          base.range=c(2000,2005)
        )
      }
    stopCluster(cl)
    

    请参阅我的使用 foreach 的并行性指南:https://privefl.github.io/blog/a-guide-to-parallelism-in-r/

    然后,要计算索引,只需使用climdex.cdd(res[[1]][[1]], spells.can.span.years = TRUE)(首先使用j,其次使用i)。

    【讨论】:

    • 谢谢,您的回答有效!不过,只需澄清一下:我应该将climdex.cdd 计算放在哪里:在 foreach 循环中还是应该为它创建一个新循环?如果我想计算除climdex.cdd 之外的另外三个索引怎么办,我应该如何将结果变量存储在循环中?很抱歉这些问题,但我在 R 中的多核处理方面没有太多经验!
    • 我认为如果新的计算速度更快,你应该创建新的循环。
    • 好的,谢谢。在处理我的实际数据时,我注意到的最后一件事是:对象res 变得非常大(大约 7 GB),因此我的服务器的 32 GB RAM 完全被这个计算所耗尽,使它变得非常慢。处理永远不会结束。有没有办法将res 写入磁盘而不是在循环期间将其保存在内存中?
    • 您始终可以将部分(例如每个 j)存储为 rds 文件,然后在您想要并行计算统计信息时加载它们(超过 j)。
    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2022-11-28
    • 2014-02-15
    • 1970-01-01
    • 2016-05-23
    • 1970-01-01
    相关资源
    最近更新 更多