PySpark读取Azure Blob中大于2GB的JSON文件失败报错

问题根源：开启multiline=true后，Spark会把整个JSON文件当作单条记录加载，而JVM整数的上限是2147483647字节（约2GB），当文件超过这个阈值时，缓冲区无法扩容就会触发IllegalArgumentException: Cannot grow BufferHolder, exceeds 2147483632 bytes错误。

以下是几种可行的解决办法：

1. 拆分大JSON文件 #

把大于2GB的JSON文件拆成多个1GB以内的小文件，保留原有嵌套结构不变。可以用Azure CLI、Python脚本或第三方工具处理Blob存储里的文件，拆分后再用原代码读取即可。

2. 转换为JSON Lines格式并关闭multiline模式 #

如果JSON结构允许，将文件转为每行一个JSON对象的JSON Lines格式（比如把原文件的features数组拆成每行一个元素），然后移除option('multiline','true')配置，让Spark按行读取，大幅降低内存压力。

3. 调整Spark内存参数（仅作备选） #

尝试调大Executor和Driver的内存以及缓冲区参数，但受限于JVM整数上限，该方法仅能解决接近2GB的文件，无法处理超过2GB的单记录加载：

# 提交Spark作业时添加参数
spark-submit \
  --executor-memory 16G \
  --driver-memory 8G \
  --conf spark.sql.buffer.size=10485760 \
  your_script.py

4. 用RDD结合流式JSON解析库处理 #

如果以上方法都无法实现，可通过RDD API结合ijson这类流式JSON解析库，分块读取并解析大文件，避免一次性加载整个文件到内存：

from pyspark.sql import Row
from azure.storage.blob import BlobClient
import ijson

def parse_large_json(blob_path):
    # 替换为你的Azure Blob连接信息
    conn_str = "your-storage-connection-string"
    container_name = "your-container-name"
    
    blob_client = BlobClient.from_connection_string(conn_str, container_name, blob_path)
    with blob_client.download_blob() as stream:
        parser = ijson.parse(stream)
        doc_type = None
        current_props = {}
        current_geom = {}
        
        for prefix, event, value in parser:
            # 捕获文档类型
            if prefix == "type" and event == "string":
                doc_type = value
            # 开始解析features.item的properties
            elif prefix == "features.item.properties" and event == "start_map":
                current_props = {}
            elif prefix.startswith("features.item.properties.") and event in ["string", "number"]:
                key = prefix.split(".")[-1]
                current_props[key] = value
            # 开始解析features.item的geometry
            elif prefix == "features.item.geometry" and event == "start_map":
                current_geom = {}
            elif prefix.startswith("features.item.geometry."):
                key = prefix.split(".")[-1]
                if event == "start_array":
                    current_geom[key] = []
                elif event == "end_array":
                    continue
                else:
                    if isinstance(current_geom.get(key), list):
                        current_geom[key].append(value)
                    else:
                        current_geom[key] = value
            # 完成一个feature的解析，生成Row
            elif prefix == "features.item" and event == "end_map":
                yield Row(type=doc_type, **current_props, json_geometry=current_geom)
                current_props = {}
                current_geom = {}

# 读取Blob文件并生成DataFrame
rdd = spark.sparkContext.parallelize([fp_in]).flatMap(parse_large_json)
df = spark.createDataFrame(rdd)

注意：需要在所有Spark Executor节点上安装ijson和azure-storage-blob库，并确保Executor有足够内存处理单条feature数据。

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