Sparklyr 配置导致 java.lang.OutOfMemoryError

Question

我在具有 8 个内核和 64Gb RAM 的本地实例上使用 R 运行 sparklyr。我的工作是 left_join 一个 [50 000 000, 12] 数据帧和一个 [20 000 000, 3] 数据帧，我使用 Spark 运行。

# Load packages
library(tidyverse)
library(sparklyr)


# Initialize configuration with defaults
config <- spark_config()

# Memory
# Set memory allocation for whole local Spark instance
# Sys.setenv("SPARK_MEM" = "50g")

# Set driver and executor memory allocations
# config$spark.driver.memory <- "8g"
# config$spark.driver.maxResultSize <- "8g"


# Connect to local cluster with custom configuration
sc <- spark_connect(master = "local", config = config, spark_home = spark_home_dir())


# Read df1 and df2
df1 <- spark_read_parquet(sc, 
                          path = "/mnt/df1/",
                          memory = FALSE, overwrite = TRUE)
df2 <- spark_read_parquet(sc, 
                          path = "/mnt/df2/",
                          memory = FALSE, overwrite = TRUE)


# Left join
df3 <- df1 %>%
  dplyr::left_join(df2)


# Write or collect
sparklyr::spark_write_parquet(df3, path="/mnt/") # or
df3 <- df3 %>% collect()

无论我如何配置 Spark 配置文件，代码都会失败并显示 java.lang.OutOfMemoryError: Java heap space。

Error: org.apache.spark.SparkException: Job aborted due to stage failure: Task 2 in stage 8.0 failed 1 times, most recent failure: Lost task 2.0 in stage 8.0 (TID 96, localhost, executor driver): java.lang.OutOfMemoryError: Java heap space

到目前为止，我已经尝试了不同的组合

Sys.setenv("SPARK_MEM" = "50g")
config["sparklyr.shell.driver-memory"] <- "20G"  
config["sparklyr.shell.num-executors"] <- 8  
config$spark.driver.maxResultSize <- "8g" 
config$spark.executor.memory <- "8g"  
config$spark.memory.fraction <- 0.9 

在 R 脚本或 spark 配置文件中。

类似的问题已经被问到1 2 3 但这些都没有解决我的问题。

Answer 1

您必须为left_join() 指定一个连接键。

否则，您将尝试计算笛卡尔积，其大小为 [1 000 000 000 000 000, 15] 并且肯定会溢出内存。

另外：避免在大型数据集上调用 collect()，因为这会将所有数据移回驱动程序，并且驱动程序很可能会出现 OOM。