R - Comment répliquer des lignes dans une structure d'étincelle à l'aide de sparklyr

Question

Existe-t-il un moyen de répliquer les lignes d'une base de données Spark en utilisant les fonctions de sparklyr/dplyr?

sc <- spark_connect(master = "spark://####:7077") 

df_tbl <- copy_to(sc, data.frame(row1 = 1:3, row2 = LETTERS[1:3]), "df") 

Ceci est la sortie désirée, enregistrée dans une nouvelle TBL d'allumage:

> df2_tbl 
    row1 row2 
    <int> <chr> 
1  1  A 
2  1  A 
3  1  A 
4  2  B 
5  2  B 
6  2  B 
7  3  C 
8  3  C 
9  3  C 

Answer 1

Avec sparklyr vous pouvez utiliser array et explode comme suggéré par @Oli:

df_tbl %>% 
    mutate(arr = explode(array(1, 1, 1))) %>% 
    select(-arr) 

# # Source: lazy query [?? x 2] 
# # Database: spark_connection 
# row1 row2 
# <int> <chr> 
# 1  1 A  
# 2  1 A  
# 3  1 A  
# 4  2 B  
# 5  2 B  
# 6  2 B  
# 7  3 C  
# 8  3 C  
# 9  3 C  

et généralisée

library(rlang) 

df_tbl %>% 
    mutate(arr = !!rlang::parse_quo(
    paste("explode(array(", paste(rep(1, 3), collapse = ","), "))") 
)) %>% select(-arr) 

# # Source: lazy query [?? x 2] 
# # Database: spark_connection 
# row1 row2 
# <int> <chr> 
# 1  1 A  
# 2  1 A  
# 3  1 A  
# 4  2 B  
# 5  2 B  
# 6  2 B  
# 7  3 C  
# 8  3 C  
# 9  3 C 

où vous pouvez facilement ajuster le nombre de lignes.

Answer 2

L'idée qui vient à l'esprit en premier est d'utiliser la fonction explode (c'est exactement ce qu'il est censé dans Étincelle). Pourtant, les tableaux ne semblent pas être pris en charge dans SparkR (au meilleur de ma connaissance).

> structField("a", "array") 
Error in checkType(type) : Unsupported type for SparkDataframe: array 

Je peux cependant proposer deux autres méthodes:

1. Un simple mais pas très élégant un:

   head(rbind(df, df, df), n=30) 
   # row1 row2 
   # 1 1 A 
   # 2 2 B 
   # 3 3 C 
   # 4 1 A 
   # 5 2 B 
   # 6 3 C 
   # 7 1 A 
   # 8 2 B 
   # 9 3 C 
   

   Ou avec une boucle pour plus généricité:

   df2 = df 
   for(i in 1:2) df2=rbind(df, df2) 
   

   Notez que ceci fonctionnerait également avec union.

2. La deuxième méthode, plus élégant (parce qu'il implique seulement une opération d'allumage) est basé sur une jointure croisée (produit cartésien) avec une trame de données de taille 3 (ou tout autre nombre):

   j <- as.DataFrame(data.frame(s=1:3)) 
   head(drop(crossJoin(df, j), "s"), n=100) 
   # row1 row2 
   # 1 1 A 
   # 2 1 A 
   # 3 1 A 
   # 4 2 B 
   # 5 2 B 
   # 6 2 B 
   # 7 3 C 
   # 8 3 C 
   # 9 3 C 
   

Answer 3

Je ne suis pas au courant d'une version côté cluster de la fonction rep de R. Nous pouvons cependant utiliser une jointure pour l'émuler côté cluster.

df_tbl <- copy_to(sc, data.frame(row1 = 1:3, row2 = LETTERS[1:3]), "df") 

replyr <- function(data, n, sc){ 
    joiner_frame <- copy_to(sc, data.frame(joiner_index = rep(1,n)), "tmp_joining_frame", overwrite = TRUE) 

    data %>% 
    mutate(joiner_index = 1) %>% 
    left_join(joiner_frame) %>% 
    select(-joiner_index) 

} 

df_tbl2 <- replyr(df_tbl, 3, sc) 
# row1 row2 
# <int> <chr> 
# 1  1 A  
# 2  1 A  
# 3  1 A  
# 4  2 B  
# 5  2 B  
# 6  2 B  
# 7  3 C  
# 8  3 C  
# 9  3 C 

Il fait le travail, mais il est un peu sale depuis le tmp_joining_frame persistera. Je ne suis pas sûr à quel point cela fonctionnera étant donné l'évaluation paresseuse sur plusieurs appels à la fonction.