Spark处理大TSV文件遇UTF16 String超限问题求解

错误原因 #

你遇到的java.lang.OutOfMemoryError: UTF16 String size is 1129672402, should be less than 1073741823不是Spark的限制，而是JVM层面的固有约束：
Java中String底层依赖char数组，每个char占2字节，单个String的char数量上限受限于JVM内存模型（错误信息中的1073741823是1GB对应的char数）。你的代码用Collectors.joining("\n")把整个1.05GB的TSV文件拼接成一个超大String，直接突破了这个限制，导致内存溢出。无法通过Spark配置增大这个阈值。

解决方案 #

方案一：用Spark内置TSV读取能力（推荐） #

Spark支持直接从HTTP/HTTPS URL读取文件，自动分布式处理，完全避免在Driver端加载全量数据：

def getGeoFeedsDataNew(spark: SparkSession, url: String, schema: StructType): DataFrame = {
    // 读取TSV文件，自动适配分布式读取
    val rawDF = spark.read
        .option("sep", "\t")
        .option("header", "true") // 若文件有表头则开启，无则设为false
        .schema(schema)
        .csv(url)

    // 执行原逻辑：每行前后添加"1"和"2"
    val processedDF = rawDF
        .withColumn("prefix", lit("1"))
        .withColumn("suffix", lit("2"))
        .select("prefix", schema.fieldNames: _*, "suffix") // 调整列顺序匹配需求

    val recordCount = processedDF.count()
    logger.info(s"records found: $recordCount")

    if (recordCount == 0) {
        spark.createDataFrame(spark.sparkContext.emptyRDD[Row], schema)
    } else {
        processedDF
    }
}

优势：利用Spark分布式计算能力，内存压力分散到集群节点，彻底避免Driver端OOM，代码简洁高效。

方案二：修改现有代码，避免生成超大String #

如果必须手动处理流，取消全量字符串拼接，直接逐行处理：

def getGeoFeedsDataNew(spark: SparkSession, url: String, schema: StructType): DataFrame = {
    val inputStream = new URL(url).openStream()
    val reader = new BufferedReader(new InputStreamReader(inputStream))

    // 逐行读取处理，跳过表头，避免拼接大字符串
    val processedLines = reader.lines()
        .skip(1) // 跳过第一行表头
        .map(line => Array("1") ++ line.split("\t") ++ Array("2"))
        .collect(Collectors.toList())

    inputStream.close()

    val recordCount = processedLines.size()
    logger.info(s"records found: $recordCount")

    if (recordCount == 0) {
        spark.createDataFrame(spark.sparkContext.emptyRDD[Row], schema)
    } else {
        val rowsRDD = spark.sparkContext.parallelize(processedLines)
            .map(row => Row.fromSeq(row.toSeq))
        spark.createDataFrame(rowsRDD, schema)
    }
}

注意：此方案仍会把全量数据加载到Driver内存，若文件持续增大，可能触发spark.driver.maxResultSize（当前配置为2G）限制，仅适合中小文件。

方案三：用RDD分布式读取文件 #

借助Spark RDD的分布式读取能力，避免Driver端加载全量数据：

def getGeoFeedsDataNew(spark: SparkSession, url: String, schema: StructType): DataFrame = {
    val linesRDD = spark.sparkContext.textFile(url)
        .zipWithIndex()
        .filter(_._2 > 0) // 跳过第一行表头
        .map(_._1)
        .map(line => Row.fromSeq(Array("1") ++ line.split("\t") ++ Array("2")))

    val recordCount = linesRDD.count()
    logger.info(s"records found: $recordCount")

    if (recordCount == 0) {
        spark.createDataFrame(spark.sparkContext.emptyRDD[Row], schema)
    } else {
        spark.createDataFrame(linesRDD, schema)
    }
}

优势：数据由集群节点分片读取处理，Driver仅负责调度，内存压力极小。

额外注意事项 #

• 确保你的schema包含新增的前缀/后缀列，或根据处理后的数据结构调整schema；
• 当前配置中spark.driver.maxResultSize=2G，若使用方案二，需保证处理后的数据量不超过该阈值。

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