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 函数来执行您的列的上下 "shifting"。它涉及创建一个新的辅助 Spark table（example2），它包含 "Numb1" 和 "Numb2" 列。然后，一旦创建了辅助 table，就可以使用 mutate

创建 "Merged" 列

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。

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

Sparklyr fill NA/NULL in spark dataframe

r

apache-spark

sparklyr