使用R sparklyr将多个Spark parquet文件加载到Spark表中？

Question

我试图使用Rsparklyr将多个parquet文件加载到一个Spark表中。附加的代码显示了我是如何做到这一点的。

spark_load_data <- function(db_conn, test_period)
{
    library(DBI)
    #
    overwrite <- TRUE
    #
    for (ts in seq(as.Date(test_period["START_DATE","VALUE"]),
                   as.Date(test_period["END_DATE","VALUE"]),
                   by="day")) {
        #
        # date to load
        #
        td <- format(as.Date(ts,origin="1970-01-01"), "%Y-%m-%d")
        #
        # load parquet files
        #
        tbl <- "pcidata"
        pq_path <- paste0("s3://<path>/PciData/transaction_date=", td)
        read_in <- spark_read_parquet(db_conn, 
                                      name=tbl,
                                      path=pq_path,
                                      overwrite=overwrite)
        #
        overwrite <- FALSE
    }
}

Answer 1

read.parquet方法实际上支持提供多个文件路径，这意味着我们可以编写一个简单的包装器：

read_parquet_multiple <- function(sc, paths) {
  spark_session(sc) %>% invoke("read") %>% invoke("parquet", as.list(paths))
}

然后使用它读取多个文件，例如（完整的示例包括连接到本地spark实例并编写2个parquet文件以供加载）：

library(sparklyr); library(dplyr)
sc <- spark_connect(master = "local")

# Write 1:10 into 2 separate parquet files
sdf_seq(sc, 1, 3, repartition = NULL) %>% spark_write_parquet("batch_1")
sdf_seq(sc, 4, 6, repartition = NULL) %>% spark_write_parquet("batch_2")

# Read mulitple files
dataset <- sc %>% read_parquet_multiple(paths = c("batch_1", "batch_2"))

# Collect to show the results
dataset %>% collect()

# # A tibble: 6 x 1
# id
# <int>
#   1     2
#   2     3
#   3     5
#   4     6
#   5     1
#   6     4