multidplyr：试用自定义功能

Question

我正在尝试通过 multidplyr::do() 在集群上学习运行自定义函数。考虑这个简单的自包含示例。例如，我正在尝试将我的自定义函数 myWxTest 应用于 flight 数据集中的每个 common_dest（具有超过 50 个航班的目的地）：

library(dplyr)
library(multidplyr)
library(nycflights13)
library(quantreg)

myWxTest <- function(x){
    stopifnot(!is.null(x$dep_time))
    stopifnot(!is.null(x$dep_delay))
    stopifnot(!is.null(x$sched_dep_time))
    stopifnot(!is.null(x$sched_arr_time))
    stopifnot(!is.null(x$arr_time))

    out_mat <- c('(Intercept)' = NA, dep_time = NA, dep_delay = NA, sched_dep_time = NA, sched_arr_time = NA)
    if(length(x$arr_time)>5){
        model_1 <- quantreg::rq(arr_time ~ dep_time + dep_delay + sched_dep_time + sched_arr_time, data = x, tau = .5)
        out_mat[names(coef(model_1))] <- coef(model_1)
    }
    return(out_mat)
}

common_dest <- flights %>%
  count(dest) %>%
  filter(n >= 365) %>%
  semi_join(flights, .) %>% 
  mutate(yday = lubridate::yday(ISOdate(year, month, day)))


cluster <- create_cluster(2)
set_default_cluster(cluster)
by_dest <- common_dest %>% 
           partition(dest, cluster = cluster)
cluster_library(by_dest, "quantreg")

到目前为止一切顺利（但我只是在重现小插图中的示例）。现在，我必须将我的自定义函数发送到每个节点：

cluster %>% cluster_call(myWxTest)

但我得到：

Error in checkForRemoteErrors(lapply(cl, recvResult)) : 
  2 nodes produced errors; first error: argument "x" is missing, with no default

最终，我想将 myWxTest 应用于每个子组：

models <- by_dest %>% 
          do(myWxTest(.))

Answer 1

我明白了运行一些调整：

library(dplyr)
library(multidplyr)
library(nycflights13)
library(quantreg)

myWxTest <- function(x){
    stopifnot(!is.null(x$dep_time))
    stopifnot(!is.null(x$dep_delay))
    stopifnot(!is.null(x$sched_dep_time))
    stopifnot(!is.null(x$sched_arr_time))
    stopifnot(!is.null(x$arr_time))

    out_mat <- c('(Intercept)' = NA, dep_time = NA, dep_delay = NA, sched_dep_time = NA, sched_arr_time = NA)
    if(length(x$arr_time)>5){
        model_1 <- quantreg::rq(arr_time ~ dep_time + dep_delay + sched_dep_time + sched_arr_time, data = x, tau = .5)
        out_mat[names(coef(model_1))] <- coef(model_1)
    }
    return(as.data.frame(out_mat, stringsAsFactors = FALSE))    # change result to data.frame, not matrix
}

common_dest <- flights %>%
    count(dest) %>%
    filter(n >= 365) %>%
    semi_join(flights, .) %>% 
    mutate(yday = lubridate::yday(ISOdate(year, month, day)))

by_dest <- common_dest %>% partition(dest)

cluster_library(by_dest, "quantreg")
cluster_copy(by_dest, myWxTest)    # copy function to each node

models <- by_dest %>% do(myWxTest(.)) %>% collect()    # collect data from clusters

...其中 returns 本地 data.frame:

models
#> Source: local data frame [390 x 2]
#> Groups: dest [78]
#> 
#>     dest     out_mat
#>    <chr>       <dbl>
#> 1    CAK 156.5248953
#> 2    CAK   0.9904261
#> 3    CAK  -0.0767928
#> 4    CAK  -0.3523211
#> 5    CAK   0.3220386
#> 6    DCA  74.5959035
#> 7    DCA   0.2751917
#> 8    DCA   1.0712483
#> 9    DCA   0.2874165
#> 10   DCA   0.4344960
#> # ... with 380 more rows

multidplyr：试用自定义功能

multidplyr: trial custom function

parallel-processing

r

dplyr

multidplyr