用Sparklyr替换NA和LOCF

Question

我的目标是用最后一次观察转发方法在星火数据框架中替换NA。我编写了以下代码并进行了工作。然而，对于一个更大的数据集，它所花费的时间似乎比预期的要长。

如果有人能推荐一种更好的方法或改进代码，那就太好了。

示例与Sparklyr代码

在下面的示例中，在使用NA对它们进行排序并按grp对其进行分组之后，将替换它们。

df_with_nas <- data.frame(time = seq(as.Date('2001/01/01'), 
                                as.Date('2010/01/01'), length.out = 10),
                          grp = c(rep(1, 5), rep(2, 5)),
                          v1 = c(1, rep(NA, 3), 5, rep(NA, 5)),
                          v2 = c(NA, NA, 3, rep(NA, 4), 3, NA, NA))

tbl <- copy_to(sc, df_with_nas, overwrite = TRUE)

tbl %>%
  spark_apply(function(df) {
    library(dplyr)
    na_locf <- function(x) {
      v <- !is.na(x)
      c(NA, x[v])[cumsum(v) + 1]
    }
    df %>% arrange(time) %>% group_by(grp) %>% mutate_at(vars(-v1, -grp), 
    funs(na_locf(.)))
  })

# # Source: spark<?> [?? x 4]
#      time   grp    v1    v2
#     <dbl> <dbl> <dbl> <dbl>
#  1 11323      1     1   NaN
#  2 11688.     1   NaN   NaN
#  3 12053.     1   NaN     3
#  4 12419.     1   NaN     3
#  5 12784.     1     5     3
#  6 13149.     2   NaN   NaN
#  7 13514.     2   NaN   NaN
#  8 13880.     2   NaN     3
#  9 14245.     2   NaN     3
# 10 14610      2   NaN     3

data.table

下面使用data.table的方法对于我所拥有的数据非常快速。我预计数据的大小很快会增加，然后我可能不得不依赖sparklyr。

library(data.table)
setDT(df_with_nas)
df_with_nas <- df_with_nas[order(time)]
cols <- c("v1", "v2")

df_with_nas[, (cols) := zoo::na.locf(.SD, na.rm = FALSE), 
                                 by = grp, .SDcols = cols]

Answer 1

我做了这种循环，很慢.

df_with_nas = df_with_nas %>%突变(row= 1:nrow(df_with_nas))

for(n in 1:50){
  df_with_nas = df_with_nas %>% 
    arrange(row) %>% 
    mutate_all(~if_else(is.na(.),lag(.,1),.))
}

跑到没有NA为止

然后

collect(df_with_nas)

会运行密码。

Answer 2

您可以利用spark_apply()函数，并在每个集群节点中运行na.locf函数。

1. 在每个集群节点上安装R运行时。
2. 在每个节点上也安装zoo R包。
3. 以这种方式运行火花应用：

data_filled <- spark_apply(data_with_holes, function(df) zoo:na.locf(df))