R -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")

到目前为止，

太好了(但我只是重现Vignette的示例(。现在，我必须将自定义功能发送到每个节点：

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

...返回本地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