R - sparklyrを使用してsparkデータフレームの行を複製する方法

Question

sparklyr/dplyrの機能を使用してSparkのデータフレームの行を複製する方法はありますか？

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

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

これは、新しいスパークTBLに保存された所望の出力、次のとおりです。（それはそれは中のためのものです正確に何である最初に思い浮かぶアイデアはexplode機能を使用することです

> 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

sparklyrを使用すると、arrayとexplodeを使用することができます。

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  

と

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 

一般

を使用すると、簡単に行数を調整することができる場所。

Answer 2

スパーク）。しかし、配列はSparkRではサポートされていないようです（私の知る限り）。

1. は簡単ではなく、非常にエレガントな1：

   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 
   

   以上のジェネリックのためのforループを持つ：

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

   私はしかし、他の二つの方法を提案することができます

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

   unionでも動作します。

   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 
   

：

秒、よりエレガントな方法は、（それが唯一の点火動作を意味するので）サイズ3のデータフレームと（デカルト積）に参加クロス（又は任意の他の数）に基づいています

Answer 3

私は、Rのrep機能のクラスタ側のバージョンを認識していません。ただし、結合を使用してクラスタ側をエミュレートすることはできます。

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 

ジョブは完了しますが、tmp_joining_frameは保持されるため、少し汚れています。関数への複数の呼び出しで怠惰な評価が与えられても、これがうまくいくかどうかはわかりません。 @Oliにより示唆されるように