如何使用R中的并行处理来分析大型时间序列数据集

我有一个大型时间序列数据集，使用顺序处理通常需要 4 小时才能处理 1800 个时间序列。我正在寻找一种使用多个核心来减少这个时间的方法，因为我有许多这样的数据集需要定期处理。

我用于顺序处理的 R 代码如下。有4个文件包含不同的数据集，每个文件包含超过1800个系列。我一直在尝试使用 doParallel 独立分析每个时间序列并将结果连接到一个文件中。即使是 CSV 文件也可以。

# load the dataset
files <- c("3MH Clean", "3MH", "6MH", "12MH")
for (j in 1:4)
{
  title <- paste("\n\n\n Evaluation of", files[j], " - Started at", date(), "\n\n\n")
  cat(title)

  History <- read.csv(paste(files[j],"csv", sep="."))

  # output forecast to XLSX
  outwb <- createWorkbook()
  sheet <- createSheet(outwb, sheetName = paste(files[j], " - ETS"))                                      
  Item <- unique(unlist(History$Item))

 for (i in 1:length(Item))
 {
    title <- paste("Evaluation of item ", Item[i], "-", i, "of", length(Item),"\n")
    cat(title)
    data <- subset(History, Item == Item[i])
    dates <- unique(unlist(data$Date))
    d <- as.Date(dates, format("%d/%m/%Y"))
    data.ts <- ts(data$Volume, frequency=12, start=c(as.numeric(format(d[1],"%Y")), as.numeric(format(d[1],"%m"))))
    try(data.ets <- ets(data.ts))
    try(forecast.ets <- forecast.ets(data.ets, h=24))
    IL <-c(Item[i],Item[i],Item[i],Item[i],Item[i],Item[i],Item[i],Item[i],Item[i],Item[i],Item[i],Item[i],Item[i],Item[i],Item[i],Item[i],Item[i],Item[i],Item[i],Item[i],Item[i],Item[i],Item[i],Item[i])
    ets.df <- data.frame(forecast.ets)
    ets.df$Item <- IL
    r <- 24*(i-1)+2
    addDataFrame(ets.df, sheet, col.names=FALSE, startRow=r)
  }

  title <- paste("\n\n\n Evaluation of", files[j], " - Completed at", date(), "\n\n\n")
  cat(title)
  saveWorkbook(outwb, paste(files[j],"xlsx",sep='.'))
}

我反映了评论中的观点，即该过程应尽可能矢量化。在没有可重现的示例的情况下，我将不得不对您的数据做出一些假设。我假设该系列一个一个地堆叠在另一个之上，并有一个变量指示该系列。如果不是，并且该系列位于单独的列中，您可以使用 reshape2 融合数据框，然后使用下面的代码。

如果您使用的是 Linux 或 Mac 机器，那么如果您能够对代码进行更多矢量化，则可以使用并行包和 mclapply。我偏爱 data.table - 如果您不熟悉，那么这可能是一个陡峭的攀登。

require(forecast)
require(data.table)
require(parallel)

#create 10 sample series -not 1800, but you get the idea.
data <- data.table(series=rep(1:10,each=60),
                   date=seq(as.Date("01/08/2009",format("%d/%m/%Y")),length.out=60,by="1 month"),
                   Volume=rep(rnorm(10),each=60)+rnorm(600))
#Define some functions to get the job done.    
getModel <- function(s) {
  data[order(date)][series==s][,ets(ts(Volume,start=as.yearmon(min(date),format("%d/%m/%Y")),frequency=12))]
}
getForecast <- function(s,forward=24) {
  model <- getModel(s)
  fc <- forecast(model,h=forward)
  return(data.frame(fc))
}
#Write the forecasts m at a time, where m is the number of cores.
Forecasts <- mclapply(1:10,getForecast)

使用数据框列表，您可以执行以下操作：

mclapply(Forecasts, function(x) {addDataFrame(x, sheet, col.names=FALSE, startRow=r))