Выполнить обнаружение точки останова (лм) параллельно в R

Question

Я делаю около 80000 вычислений для определения точек останова временных рядов в R. У меня есть все эти чрезвычайно разные временные ряды, где я не могу применять модели ARIMA, поэтому я рассчитываю линейную модель для временных рядов, затем извлекаю точки останова и использую подогнанные результаты регрессии. рассчитать тренд, исходящий из последней точки останова.

В приведенном выше примере алгоритм обнаружит три точки останова (один наклон, один довольно плоский и один спад). Это идеально подходит для моих нужд, но выполнение вычислений 80k точек останова один раз в неделю - это слишком большие затраты, поэтому я пытаюсь реализовать это с помощью параллельной обработки в R.

В этом примере (найти ссылку на данные ниже) я последовательно вычисляю контрольные точки, что занимает около 24 часов для всех 88k.

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 времени.

Ниже я перечислил три способа, которыми я пытался распараллелить вычисления, но я не могу получить вложенное применение для работы в параллельной настройке.

С параллельным пакетом

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

Я ценю ваши идеи и отзывы!

Answer 1

Задавая вопрос и описывая проблему, вы решите его большую часть времени... Я понял, что mutate не работает нигде (поверьте мне, Stackru) параллельно в R.

Поэтому я перешел на использование do и распределяю нагрузку через multidplyr и получаю сокращение времени вычислений примерно на 50% при переходе от 1 до 4 ядер и до 25% общего времени при переходе от 1 до 8 ядер.

Код ниже.

## 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])) .