r - Sparklyr 的 spark_apply 函数似乎在单个执行程序上运行并且在中等大型数据集上失败

Question

我正在尝试使用 spark_apply 在 Spark 表上运行下面的 R 函数。如果我的输入表很小(例如 5,000 行)，这可以正常工作，但是当表中等大(例如 5,000,000 行)时，大约 30 分钟后会抛出错误:sparklyr worker rscript 失败，检查 worker 日志以获取详细信息

查看 Spark UI 会发现只有一个任务被创建，并且一个执行器被应用到这个任务。

任何人都可以就为什么这个函数对于 500 万行数据集失败提出建议吗？问题可能是让一个执行程序完成所有工作，但失败了吗？

# Create data and copy to Spark
testdf <- data.frame(string_id=rep(letters[1:5], times=1000), # 5000 row table
                 string_categories=rep(c("", "1", "2 3", "4 5 6", "7"), times=1000))
testtbl <- sdf_copy_to(sc, testdf, overwrite=TRUE, repartition=100L, memory=TRUE)

# Write function to return dataframe with strings split out
myFunction <- function(inputdf){
  inputdf$string_categories <- as.character(inputdf$string_categories)
  inputdf$string_categories=with(inputdf, ifelse(string_categories=="", "blank", string_categories))
  stringCategoriesList <- strsplit(inputdf$string_categories, ' ')
  outDF <- data.frame(string_id=rep(inputdf$string_id, times=unlist(lapply(stringCategoriesList, length))),
                  string_categories=unlist(stringCategoriesList))
 return(outDF)
}

# Use spark_apply to run function in Spark
outtbl <- testtbl %>%
  spark_apply(myFunction,
          names=c('string_id', 'string_categories'))
outtbl

Answer 1

1. sparklyr worker rscript failure, check worker logs for details error 由 driver 节点写入并指出实际错误需要在 worker 日志中找到。通常，可以通过打开 stdout 来访问工作日志。从 Spark UI 中的执行者选项卡；日志应包含 RScript:描述执行程序正在处理的内容以及错误的具体情况的条目。

2. 关于正在创建的单个任务，当columns未指定 spark_apply() 中的类型，它需要计算结果的一个子集来猜测列类型，为避免这种情况，按如下方式传递显式列类型:

   outtbl <- testtbl %>%
spark_apply(
myFunction,
columns=list(
string_id = "character",
string_categories = "character"))

3. 如果使用 sparklyr 0.6.3 , 更新到 sparklyr 0.6.4或 devtools::install_github("rstudio/sparklyr") , 自 sparklyr 0.6.3在启用包分发且每个节点中运行多个执行程序的某些边缘情况下包含不正确的等待时间。

4. 在高负载下，内存不足是很常见的。增加分区数可以解决此问题，因为它会减少处理此数据集所需的总内存。尝试将其运行为:

   testtbl %>%
sdf_repartition(1000) %>%
spark_apply(myFunction, names=c('string_id', 'string_categories'))

5. 也可能是由于函数中的逻辑，函数对某些分区抛出异常，您可以使用 tryCatch() 查看是否属于这种情况。忽略错误，然后找出哪些是缺失值，以及为什么函数会因这些值而失败。我将从以下内容开始:

   myFunction <- function(inputdf){
tryCatch({
inputdf$string_categories <- as.character(inputdf$string_categories)
inputdf$string_categories=with(inputdf, ifelse(string_categories=="", "blank", string_categories))
stringCategoriesList <- strsplit(inputdf$string_categories, ' ')
outDF <- data.frame(string_id=rep(inputdf$string_id, times=unlist(lapply(stringCategoriesList, length))),
string_categories=unlist(stringCategoriesList))
return(outDF)
}, error = function(e) {
return(
data.frame(string_id = c(0), string_categories = c("error"))
)
})
}