R:使用 foreach 读取 csv 数据并对数据应用函数并导出回 csv

Question

我有3个csv文件，分别是file1.csv，file2.csv和file3.csv。

现在对于每个文件，我想导入 csv 并对其执行一些功能，然后导出转换后的 csv。因此，3 个 csv 输入和 3 个转换 csv 输出。而且只有 3 个独立的任务。所以我想我可以尝试使用foreach%dopar%.请不要说我使用的是窗口机器。

但是，我无法让它工作。

library(foreach)
library(doParallel)
library(xts)
library(zoo)
numCores <- detectCores()
cl <- parallel::makeCluster(numCores)
doParallel::registerDoParallel(cl)
filenames <- c("file1.csv","file2.csv","file3.csv")
foreach(i = 1:3, .packages = c("xts","zoo")) %dopar%{
df_xts          <- data_processing_IMPORT(filenames[i])
ddates                <- unique(date(df_xts))
}

如果我注释掉ddates <- unique(date(df_xts))的最后一行，代码运行良好，没有错误。

但是，如果我包含最后一行代码，我会收到以下错误，我不知道要解决。我试图添加.export = c("df_xts").

Error in { : task 1 failed - "unused argument (df_xts)"

它仍然不起作用。我想了解我的逻辑出了什么问题，我应该如何解决这个问题？我只是尝试仅对数据应用简单的函数，我仍然没有转换数据并将它们单独导出为 csv。然而我已经卡住了。

有趣的是，我已经编写了下面的简单代码，效果很好。在foreach中，a就像上面的df_xts一样，被存储在一个变量中，并传递到Fun2进行处理。下面的代码工作正常。但以上没有。我不明白为什么。

numCores <- detectCores()
cl <- parallel::makeCluster(numCores)
doParallel::registerDoParallel(cl)

# Define the function
Fun1=function(x){
a=2*x
b=3*x
c=a+b
return(c)
}
Fun2=function(x){
a=2*x
b=3*x
c=a+b
return(c)
}
foreach(i = 1:10)%dopar%{
x <- rnorm(5)
a <- Fun1(x)
tst <- Fun2(a)
return(tst)
}
### Output: No error
parallel::stopCluster(cl)

更新：我发现问题出在提取csv文件中日期数的date功能上，但我不确定如何解决这个问题。

Answer 1

foreach()的使用是正确的。您在ddates <- unique(date(df_xts))中使用了date()，但此函数以 POSIX 的形式返回当前系统时间，并且不需要任何参数。因此，参数错误与date()函数有关。

所以我想你想用as.Date()或类似的东西代替。

ddates <- unique(as.Date(df_xts))

Answer 2

我在读取，修改和写入多个CSV文件时遇到了同样的问题。我试图为此找到一个tidyverse的解决方案，虽然它并没有真正解决上面的date问题，但它是 - 如何使用purrrmap读取，修改和写入多个csv文件。

library(tidyverse)

# There are some sample csv file in the "sample" dir.
# First get the paths of those.
datapath <- fs::dir_ls("./sample", regexp = ("csv"))
datapath
# Then read in the data, such as it is a list of data frames
# It seems simpler to write them back to disk as separate files.
# Another way to read them would be:
# newsampledata <- vroom::vroom(datapath, ";", id = "path")
# but this will return a DF and separating it to different files
# may be more complicated.
sampledata <- map(datapath, ~ read_delim(.x, ";"))
# Do some transformation of the data.
# Here I just alter the column names.
transformeddata <- sampledata %>% 
map(rename_all, tolower)
# Then prepare to write new files
names(transformeddata) <- paste0("new-", basename(names(transformeddata)))
# Write the csv files and check if they are there
map2(transformeddata, names(transformeddata), ~ write.csv(.x, file = .y))
dir(pattern = "new-")