Sparklyr 在 spark 数据框中填充 NA/NULL

Question

我想将 spark 数据帧中的 NA/NULL 分配给它最近的邻居。我来自 R 背景，所以我使用 sparklyr 但想不出办法。

这是一个示例代码：

 set.seed(1)    
 example <- data.frame (ID = 1:10, Cat = letters[1:5], 
                       Numb = sample(c(NA, NA, NA, NA, 1:10), 10))
     ID Cat Numb
  1   1   a   NA
  2   2   b    1
  3   3   c    3
  4   4   d    6
  5   5   e   NA
  6   6   a    5
  7   7   b    4
  8   8   c    9
  9   9   d   10
  10 10   e   NA 

所以我想填写 Numb 列，ID 1 NA 到 ID2 Numb 1，ID 5 到 id 4 和 6（6 或 5），以及 ID 10 到 ID 9 值 (10)。它可以很容易地在 R 中完成。无论如何可以通过 Sparklyr 在 Spark 中执行此操作吗？

这是我的 R 解决方案：

example$Numb1 <- example$Numb[c(1,1:(nrow(example)-1))]
example$Numb2 <- example$Numb[c(2:(nrow(example)), nrow(example))]
example$Merge <- ifelse(is.na(example$Numb), ifelse(is.na(example$Numb1), 
example$Numb2, example$Numb1), example$Numb)

    ID Cat Numb Numb1 Numb2 Merge
1   1   a   NA    NA     1     1
2   2   b    1    NA     3     1
3   3   c    3     1     6     3
4   4   d    6     3    NA     6
5   5   e   NA     6     5     6
6   6   a    5    NA     4     5
7   7   b    4     5     9     4
8   8   c    9     4    10     9
9   9   d   10     9    NA    10
10 10   e   NA    10    NA    10

当然，如果我在连续的行中有多个 NA 值，事情会变得更加复杂。也许可以提出另一个建议。

但对于 sparklyr，我不知道我能做什么。

Answer 1

这是一个部分工作的解决方案，其中包含一个 SQL 查询和一个来自 dplyr 包的 mutate 函数。它没有解决连续行中多个 NA 值的情况，因为它是您的基本 R 解决方案的翻译，但它可能对其他（更完整的）方法有用。

我使用了 HiveQL 的 Lag and Lead 函数来执行列的上下“移动”。它涉及创建一个新的辅助 Spark 表 (example2)，其中包含“Numb1”和“Numb2”列。然后，一旦创建了辅助表，您就可以使用mutate 创建“合并”列

library(DBI)
library(sparklyr)
library(dplyr)

set.seed(1)    
exampleDF <- data.frame (ID = 1:10, Cat = letters[1:5], 
                         Numb = sample(c(NA, NA, NA, NA, 1:10), 10))

# Connection to Spark and creation of the table to test.
sc <- spark_connect("local")
example <- copy_to(sc, exampleDF)  

# Create a Spark table with columns Numb1 and Numb2
DBI::dbSendQuery(sc, "CREATE TABLE example2 AS (SELECT ID, Cat, Numb, LAG(Numb, 1) over (PARTITION BY 1 ORDER BY ID) AS Numb1,
             LEAD(Numb, 1) over (PARTITION BY 1 ORDER BY ID) AS Numb2 FROM exampledf)")

# Load the auxiliary table as a Spark DataFrame
ex2 <- tbl(sc, "example2")

# Mutate in order to create the Merged column
res <- ex2 %>%
  mutate(Merged = ifelse(is.na(Numb), ifelse(is.na(Numb1), Numb2, Numb1), Numb))

res

# Source:   lazy query [?? x 6]
# Database: spark_connection
      id   cat  numb numb1 numb2 Merged
   <int> <chr> <int> <int> <int>  <int>
 1     1     a    NA    NA     1      1
 2     2     b     1    NA     3      1
 3     3     c     3     1     6      3
 4     4     d     6     3    NA      6
 5     5     e    NA     6     5      6
 6     6     a     5    NA     4      5
 7     7     b     4     5     9      4
 8     8     c     9     4    10      9
 9     9     d    10     9    NA     10
10    10     e    NA    10    NA     10

附带说明，您还可以通过COALESCE 函数避免使用mutate 函数（以及所有ifelses）。我认为这样会更有效率。

DBI::dbGetQuery(sc, "SELECT ID, Cat, Numb, COALESCE(Numb, Numb1, Numb2) AS Merged FROM example2")
   ID Cat Numb Merged
1   1   a   NA      1
2   2   b    1      1
3   3   c    3      3
4   4   d    6      6
5   5   e   NA      6
6   6   a    5      5
7   7   b    4      4
8   8   c    9      9
9   9   d   10     10
10 10   e   NA     10

我希望这会有所帮助。

已编辑

如果您想完全避免使用 SQL，您也可以使用 dplyr 函数：

example %>% arrange(ID) %>%
    mutate(Numb1 = lag(Numb, 1)) %>%
    mutate(Numb2 = lead(Numb, 1L)) %>%
    mutate(Merged = ifelse(is.na(Numb), ifelse(is.na(Numb1), Numb2, Numb1), Numb))
# Source:     lazy query [?? x 6]
# Database:   spark_connection
# Ordered by: ID
      ID   Cat  Numb Numb1 Numb2 Merged
   <int> <chr> <int> <int> <int>  <int>
 1     1     a    NA    NA     1      1
 2     2     b     1    NA     3      1
 3     3     c     3     1     6      3
 4     4     d     6     3    NA      6
 5     5     e    NA     6     5      6
 6     6     a     5    NA     4      5
 7     7     b     4     5     9      4
 8     8     c     9     4    10      9
 9     9     d    10     9    NA     10
10    10     e    NA    10    NA     10
# ... with more rows

我在编码两个连续的mutate 函数时遇到了一些麻烦（这就是我首先使用混合 SQL-dplyr 方法的原因）。我最终在 sparklyr 上开了一个issue。