Databricks中SparkR无法从1600万行R data.frame创建Spark DataFrame

问题核心 #

1600万条记录的本地R DataFrame直接通过createDataFrame转换为Spark DataFrame时，因单次数据序列化、传输开销过大触发Rserve超时，调整分区数无法解决本质问题。

可行解决方案 #

1. 分批次转换合并 #

将本地数据拆分为小批次逐个转换，降低单次传输的数据量，同时及时清理临时变量释放内存：

sparkR.session()

# 设置每批次处理100万条记录，可根据内存情况调整
batch_size <- 1000000
total_rows <- nrow(local_r_data_frame)
num_batches <- ceiling(total_rows / batch_size)

spark_df <- NULL

for (i in 1:num_batches) {
  start <- (i - 1) * batch_size + 1
  end <- min(i * batch_size, total_rows)
  batch <- local_r_data_frame[start:end, ]
  
  # 转换当前批次
  batch_spark <- createDataFrame(batch)
  
  # 合并到主Spark DataFrame
  if (is.null(spark_df)) {
    spark_df <- batch_spark
  } else {
    spark_df <- union(spark_df, batch_spark)
  }
  
  # 清理临时变量
  rm(batch, batch_spark)
  gc()
}

2. 先写入文件再读取 #

将R DataFrame写入高效格式（如Parquet），再通过Spark读取，绕过直接序列化传输的瓶颈：

sparkR.session()

# 写入Databricks本地临时路径的Parquet文件
write_parquet(local_r_data_frame, "/tmp/r_local_data.parquet")

# Spark读取文件
spark_df <- read.parquet("/tmp/r_local_data.parquet")

# 可选：清理临时文件
unlink("/tmp/r_local_data.parquet", recursive = TRUE)

3. 调整Rserve超时配置 #

在Databricks集群的Spark配置中增加Rserve超时时间：

• 进入集群配置页面，添加以下配置：

  spark.r.rserve.timeout 3600
  

  将超时时间设为3600秒（1小时），可根据实际转换时长调整。

4. 优化本地数据内存占用 #

压缩数据体积，减少序列化开销：

# 将字符串列转为因子（如果不需要保留完整字符串特性）
local_r_data_frame$string_col <- as.factor(local_r_data_frame$string_col)

# 将数值列转为更小类型（如double转integer，需确保数值范围符合）
local_r_data_frame$num_col <- as.integer(local_r_data_frame$num_col)

内容的提问来源于stack exchange，提问作者Sanchez333