使用 Sparklyr 将 NA 替换为 LOCF答案

【问题标题】：Replacing NA's with LOCF using Sparklyr使用 Sparklyr 将 NA 替换为 LOCF
【发布时间】：2019-11-14 03:06:26
【问题描述】：

我的目标是使用 Last Observation Carried Forward 方法替换 spark 数据框中的 NA。我编写了以下代码并工作。但是，对于更大的数据集，它似乎需要比预期更长的时间。

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

使用 Sparklyr 的示例和代码

在以下示例中，NA 在使用 time 并按 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]

【问题讨论】：

标签： r apache-spark sparklyr

【解决方案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)

将运行代码。

【讨论】：

【解决方案2】：

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

在每个集群节点上安装 R 运行时。
也在每个节点上安装zoo R 包。
运行 spark 应用这种方式：

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

【讨论】：

【解决方案3】：

您可以使用 sql 快速完成此操作，另外还有一个好处是您可以轻松地在分组的基础上应用 LOCF。您要使用的模式是 LAST_VALUE(column, true) OVER (window) - 这会在窗口中搜索不是 NA 的最新列值（将“true”传递给 LAST_VALUE 设置 ignore NA = true）。由于您想从当前值向后看，因此窗口应该是

ORDER BY time
ROWS BETWEEN UNBOUNDED PRECEDING AND -1 FOLLOWING

当然，如果组中的第一个值是 NA，它将保持 NA。

library(sparklyr)
library(dplyr)

sc <- spark_connect(master = "local")

test_table <- data.frame(
  v1 =   c(1, 2, NA, 3, NA, 5, NA, 6, NA),
  v2 =   c(1, 1,  1, 1,  1, 2,  2, 2,  2),
  time = c(1, 2,  3, 4,  5, 2,  1, 3,  4)
) %>%
  sdf_copy_to(sc, ., "test_table")

spark_session(sc) %>%
  sparklyr::invoke("sql", "SELECT *, LAST_VALUE(v1, true)
                   OVER (PARTITION BY v2
                         ORDER BY time
                         ROWS BETWEEN UNBOUNDED PRECEDING AND -1 FOLLOWING) 
                   AS last_non_na
                   FROM test_table") %>%
  sdf_register() %>%
  mutate(v1 = ifelse(is.na(v1), last_non_na, v1))
#> # Source: spark<?> [?? x 4]
#>      v1    v2  time last_non_na
#>   <dbl> <dbl> <dbl>       <dbl>
#> 1     1     1     1         NaN
#> 2     2     1     2           1
#> 3     2     1     3           2
#> 4     3     1     4           2
#> 5     3     1     5           3
#> 6   NaN     2     1         NaN
#> 7     5     2     2         NaN
#> 8     6     2     3           5
#> 9     6     2     4           6

^{由reprex package (v0.3.0) 于 2019 年 8 月 27 日创建}

【讨论】：