分块拉取Blob存储大CSV文件接收HTTP响应时报java.lang.OutOfMemoryError

AI 小北

2026-10-13

问题原因分析

DataFrame不可变特性导致的内存累积：Spark的DataFrame是immutable结构，每次调用union方法都会生成一个新的DataFrame，旧的DataFrame并不会立即被垃圾回收，同时逻辑执行计划会随着循环次数增加持续膨胀，所有历史拉取的500MB数据块都会被保留在内存中等待计算，4次累计就超过了堆内存上限。
大字符串对象占用堆内存未及时释放：你每次拉取的500MB分段会被mkString("\n")拼接为一个完整的String对象存储在堆内存中，这类大对象会直接进入JVM的老年代，不会被young GC快速回收，在DataFrame的引用还存在的情况下，这些大字符串会一直占用堆内存。
潜在的Range请求配置缺失：从你给出的代码来看，循环中调用GetResponse时传入的headers没有动态设置Range字段，如果没有每次指定bytes=start-end的范围，你每次请求都会拉取完整的6GB文件，而不是500MB分段，这会直接导致内存快速耗尽。
Source.fromInputStream的资源泄漏风险：你虽然关闭了InputStream，但Source.fromInputStream本身也需要手动关闭，否则其内部持有的缓冲区资源也会占用内存。

解决方案

避免循环union DataFrame，改为批量写入存储
不要在循环中反复union DataFrame，你可以将每次拉取的分段直接写入临时磁盘存储（比如本地磁盘、HDFS、Blob存储的临时目录），所有分段拉取完成后再一次性读取生成DataFrame，避免全量数据在内存中累积。
参考代码：

// 每次拉取后直接写入临时路径，不要在内存中union
for(i <- 1 to chunkNum) {
  val response = GetResponse(headers, "https://myblobstorage.blob.core.windows.net/test/test6gb.csv")
  Seq((response, currentDate)).toDF("Chunk", "InsertedDate")
    .write.mode("append").parquet("tmp/blob_chunks/")
}
// 全部拉完后读取合并
val dfRestAPI = spark.read.parquet("tmp/blob_chunks/")

优化大对象内存占用
不要将500MB的完整内容拼接为单个String，你可以直接将InputStream读取为字节数组写入存储，或者按行拆分后分批处理，避免生成超大字符串对象。同时调整JVM堆内存配置，增加-Xmx参数设置，开启大对象GC优化。
补充分段请求逻辑
在循环中动态构造Range请求头，确保每次只拉取对应分段的数据：

val chunkSize = 500 * 1024 * 1024 // 500MB
for(i <- 0 until chunkNum) {
  val start = i * chunkSize
  val end = Math.min((i+1)*chunkSize - 1, totalFileSize - 1)
  val headersWithRange = headers + ("Range" -> s"bytes=$start-$end")
  val response = GetResponse(headersWithRange, url)
  // 后续处理逻辑
}

修复资源泄漏问题
修改GetResponse函数，手动关闭Source对象：

def GetResponse(headers: Map[String, String], url: String): String = {
    val obj: URL = new URL(url)
    val con: HttpsURLConnection = obj.openConnection().asInstanceOf[HttpsURLConnection]
    var iStream: InputStream = null
    var source: scala.io.Source = null
    try {
      var output: String = null
      con.setRequestMethod("GET")
      headers.foreach { case (k, v) => con.setRequestProperty(k, v) }
      val httpResponseCode = con.getResponseCode
      if (httpResponseCode == 200 || httpResponseCode == 206) {
        iStream = con.getInputStream
        source = Source.fromInputStream(iStream)
        output = source.getLines().mkString("\n")
      }
      output
    } catch {
      case exception: Exception => throw new Exception("Http Handler exception in GetHttpResponse", exception)
    } finally {
      if(source != null) source.close()
      if(iStream != null) iStream.close()
      con.disconnect()
    }
}

开启Spark动态资源释放配置
设置Spark参数spark.sql.shuffle.partitions为合理值，开启spark.cleaner.referenceTracking.cleanCheckpoints、spark.cleaner.periodicGC.interval参数，让Spark定期回收无用的内存资源。

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