2

我在 R 中进行了大约 80000 个时间序列断点检测计算。我有所有这些极其不同的时间序列，我无法应用 ARIMA 模型，所以我正在计算每个时间序列的线性模型，然后提取断点并使用回归的拟合结果计算来自最后一个断点的趋势。
一个时间序列的示例

在上面的示例中，算法将检测三个断点（一个倾斜、一个相当平坦和一个下降）。它非常适合我的需求，但是每周按顺序运行一次 80k 断点计算开销太大，因此我试图通过在 R 中使用并行处理来实现这一点。

在此示例中（在下面找到指向数据的链接）我正在按顺序计算断点，所有 88k 大约需要 24 小时。

df.subset <- read.csv("dfsubset.csv)"
start <- Sys.time()

All.Breakpoints <- df.subset %>%
nest(-CONC_ID) %>%
mutate(bps = map(data, ~breakpoints(ACT_QTY_new ~ Index, data = .)))

Sys.time() - start

在这个代码片段中，我正在对 10 个时间序列（在我的 Mac 上）运行检测，这需要47 秒。我猜想并行化应该将这个基准测试时间减少到大约 1/4 pf 时间。

下面我列出了三种我尝试并行计算的方法，但我无法让嵌套应用在并行设置中工作。

与并行包

clus <- makeCluster(4)
clusterEvalQ(clus, {library(dplyr); library(tidyr); library(magrittr)})

myfunction <- function(df.subset) {
All.Breakpoints <- df.subset %>%
nest(-CONC_ID) %>%
mutate(bps = map(data, ~breakpoints(ACT_QTY_new ~ Index, data = .)))
return(All.Breakpoints)
}

clusterExport(clus, "myfunction")

do.call(bind_rows, parApply(clus, df.subset, 1,{function(r) { 
myfunction(r[1]) }}))

使用 multidplyr 包：

library(multidplyr)
cluster <- create_cluster(4)
set_default_cluster(cluster)

four <- function(x) {
All.Breakpoints <- x %>%
nest(-CONC_ID) %>%
mutate(bps = map(data, ~breakpoints(ACT_QTY_new ~ Index, data = .)))
return(All.Breakpoints)
}

cluster_assign_value(cluster, 'four', four)
save <- df.subset %>% partition(CONC_ID) %>% map(four(.))

与并行包但其他分组

library(parallel)
cl <- detectCores()

group <- df.subset %>% group_by(CONC_ID) %>% group_indices
df.subset <- bind_cols(tibble(group), df.subset)

cluster <- create_cluster(cores = cl)

by_group <- df.subset %>%
partition(group, cluster = cluster)

by_group %>%
# Assign libraries
cluster_library("tidyr") %>%
cluster_library("dplyr") %>%
cluster_library("strucchange") %>%
cluster_library("purrr") %>%
# Assign values (use this to load functions or data to each core)
cluster_assign_value("df.subset", df.subset) 

cluster_eval(by_group, search())[[1]] # results for first cluster shown 
only
cluster_get(by_group, "df.subset")[[1]]

start <- proc.time() # Start clock
sp_500_processed_in_parallel <- by_group %>% # Use by_group party_df
group_by(CONC_ID) %>%
mutate(bps = map(data, ~breakpoints(ACT_QTY_new ~ Index, data = .))) 
%>%
collect() %>% # Special collect() function to recombine partitions
as_tibble()   # Convert to tibble
time_elapsed_parallel <- proc.time() - start # End clock
time_elapsed_parallel

文件链接：

http://www.filedropper.com/dfsubset

感谢您的想法和反馈！

4

1 回答 1

0

提出问题并描述问题将在大多数情况下为您自己解决...我发现 mutate在 R 中并行不工作（老实说，Stackoverflow ）。

因此，我改为使用do并通过multidplyr分配负载，从 1 核到 4 核时计算时间减少了约 50%，从 1 核到 8 核时减少了总时间的 25%。

代码如下。

## parallel
cl <- detectCores()
cl

df.cluster <- df.subset

cluster <- create_cluster(cores = cl)
cluster

by_group <- df.cluster %>%
partition(CONC_ID, cluster = cluster)
by_group

by_group %>%

# Assign libraries
cluster_library("strucchange")
cluster_eval(by_group, search())[[1]] # results for first cluster shown only

start <- proc.time() # Start clock

cluster.processed <- by_group %>%
                     do(model = breakpoints(ACT_QTY_new ~ Index, data = .)) %>%
                     collect()

time_elapsed_parallel <- proc.time() - start # End clock
time_elapsed_parallel

rm(by_grou)
gc()

Predictions <- cluster.processed %>%
mutate(SegmentedForecast = map(model, fitted))
df.fitted.vector <- as.data.frame(rowwise(Predictions[,3])) .
于 2018-10-12T12:39:09.520 回答